lustre/obdclass/lu_object.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * 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 #ifndef EXPORT_SYMTAB
  47 # define EXPORT_SYMTAB
  48 #endif
  49
  50 #include <libcfs/libcfs.h>
  51
  52 #ifdef __KERNEL__
  53 # include <linux/module.h>
  54 #endif
  55
  56 /* hash_long() */
  57 #include <libcfs/libcfs_hash.h>
  58 #include <obd_class.h>
  59 #include <obd_support.h>
  60 #include <lustre_disk.h>
  61 #include <lustre_fid.h>
  62 #include <lu_object.h>
  63 #include <libcfs/list.h>
  64 /* lu_time_global_{init,fini}() */
  65 #include <lu_time.h>
  66
  67 static void lu_object_free(const struct lu_env *env, struct lu_object *o);
  68
  69 /**
  70  * Decrease reference counter on object. If last reference is freed, return
  71  * object to the cache, unless lu_object_is_dying(o) holds. In the latter
  72  * case, free object immediately.
  73  */
  74 void lu_object_put(const struct lu_env *env, struct lu_object *o)
  75 {
  76         struct lu_object_header *top;
  77         struct lu_site          *site;
  78         struct lu_object        *orig;
  79         int                      kill_it;
  80
  81         top = o->lo_header;
  82         site = o->lo_dev->ld_site;
  83         orig = o;
  84         kill_it = 0;
  85         write_lock(&site->ls_guard);
  86         if (atomic_dec_and_test(&top->loh_ref)) {
  87                 /*
  88                  * When last reference is released, iterate over object
  89                  * layers, and notify them that object is no longer busy.
  90                  */
  91                 list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
  92                         if (o->lo_ops->loo_object_release != NULL)
  93                                 o->lo_ops->loo_object_release(env, o);
  94                 }
  95                 -- site->ls_busy;
  96                 if (lu_object_is_dying(top)) {
  97                         /*
  98                          * If object is dying (will not be cached), removed it
  99                          * from hash table and LRU.
 100                          *
 101                          * This is done with hash table and LRU lists
 102                          * locked. As the only way to acquire first reference
 103                          * to previously unreferenced object is through
 104                          * hash-table lookup (lu_object_find()), or LRU
 105                          * scanning (lu_site_purge()), that are done under
 106                          * hash-table and LRU lock, no race with concurrent
 107                          * object lookup is possible and we can safely destroy
 108                          * object below.
 109                          */
 110                         hlist_del_init(&top->loh_hash);
 111                         list_del_init(&top->loh_lru);
 112                         -- site->ls_total;
 113                         kill_it = 1;
 114                 }
 115         }
 116         write_unlock(&site->ls_guard);
 117         if (kill_it)
 118                 /*
 119                  * Object was already removed from hash and lru above, can
 120                  * kill it.
 121                  */
 122                 lu_object_free(env, orig);
 123 }
 124 EXPORT_SYMBOL(lu_object_put);
 125
 126 /**
 127  * Allocate new object.
 128  *
 129  * This follows object creation protocol, described in the comment within
 130  * struct lu_device_operations definition.
 131  */
 132 static struct lu_object *lu_object_alloc(const struct lu_env *env,
 133                                          struct lu_device *dev,
 134                                          const struct lu_fid *f,
 135                                          const struct lu_object_conf *conf)
 136 {
 137         struct lu_object *scan;
 138         struct lu_object *top;
 139         struct list_head *layers;
 140         int clean;
 141         int result;
 142         ENTRY;
 143
 144         /*
 145          * Create top-level object slice. This will also create
 146          * lu_object_header.
 147          */
 148         top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
 149         if (top == NULL)
 150                 RETURN(ERR_PTR(-ENOMEM));
 151         /*
 152          * This is the only place where object fid is assigned. It's constant
 153          * after this point.
 154          */
 155         LASSERT(fid_is_igif(f) || fid_ver(f) == 0);
 156         top->lo_header->loh_fid  = *f;
 157         layers = &top->lo_header->loh_layers;
 158         do {
 159                 /*
 160                  * Call ->loo_object_init() repeatedly, until no more new
 161                  * object slices are created.
 162                  */
 163                 clean = 1;
 164                 list_for_each_entry(scan, layers, lo_linkage) {
 165                         if (scan->lo_flags & LU_OBJECT_ALLOCATED)
 166                                 continue;
 167                         clean = 0;
 168                         scan->lo_header = top->lo_header;
 169                         result = scan->lo_ops->loo_object_init(env, scan, conf);
 170                         if (result != 0) {
 171                                 lu_object_free(env, top);
 172                                 RETURN(ERR_PTR(result));
 173                         }
 174                         scan->lo_flags |= LU_OBJECT_ALLOCATED;
 175                 }
 176         } while (!clean);
 177
 178         list_for_each_entry_reverse(scan, layers, lo_linkage) {
 179                 if (scan->lo_ops->loo_object_start != NULL) {
 180                         result = scan->lo_ops->loo_object_start(env, scan);
 181                         if (result != 0) {
 182                                 lu_object_free(env, top);
 183                                 RETURN(ERR_PTR(result));
 184                         }
 185                 }
 186         }
 187
 188         dev->ld_site->ls_stats.s_created ++;
 189         RETURN(top);
 190 }
 191
 192 /**
 193  * Free an object.
 194  */
 195 static void lu_object_free(const struct lu_env *env, struct lu_object *o)
 196 {
 197         struct list_head splice;
 198         struct lu_object *scan;
 199         struct lu_site          *site;
 200         struct list_head        *layers;
 201
 202         site   = o->lo_dev->ld_site;
 203         layers = &o->lo_header->loh_layers;
 204         /*
 205          * First call ->loo_object_delete() method to release all resources.
 206          */
 207         list_for_each_entry_reverse(scan, layers, lo_linkage) {
 208                 if (scan->lo_ops->loo_object_delete != NULL)
 209                         scan->lo_ops->loo_object_delete(env, scan);
 210         }
 211
 212         /*
 213          * Then, splice object layers into stand-alone list, and call
 214          * ->loo_object_free() on all layers to free memory. Splice is
 215          * necessary, because lu_object_header is freed together with the
 216          * top-level slice.
 217          */
 218         CFS_INIT_LIST_HEAD(&splice);
 219         list_splice_init(layers, &splice);
 220         while (!list_empty(&splice)) {
 221                 /*
 222                  * Free layers in bottom-to-top order, so that object header
 223                  * lives as long as possible and ->loo_object_free() methods
 224                  * can look at its contents.
 225                  */
 226                 o = container_of0(splice.prev, struct lu_object, lo_linkage);
 227                 list_del_init(&o->lo_linkage);
 228                 LASSERT(o->lo_ops->loo_object_free != NULL);
 229                 o->lo_ops->loo_object_free(env, o);
 230         }
 231         cfs_waitq_broadcast(&site->ls_marche_funebre);
 232 }
 233
 234 /**
 235  * Free \a nr objects from the cold end of the site LRU list.
 236  */
 237 int lu_site_purge(const struct lu_env *env, struct lu_site *s, int nr)
 238 {
 239         struct list_head         dispose;
 240         struct lu_object_header *h;
 241         struct lu_object_header *temp;
 242
 243         CFS_INIT_LIST_HEAD(&dispose);
 244         /*
 245          * Under LRU list lock, scan LRU list and move unreferenced objects to
 246          * the dispose list, removing them from LRU and hash table.
 247          */
 248         write_lock(&s->ls_guard);
 249         list_for_each_entry_safe(h, temp, &s->ls_lru, loh_lru) {
 250                 /*
 251                  * Objects are sorted in lru order, and "busy" objects (ones
 252                  * with h->loh_ref > 0) naturally tend to live near hot end
 253                  * that we scan last. Unfortunately, sites usually have small
 254                  * (less then ten) number of busy yet rarely accessed objects
 255                  * (some global objects, accessed directly through pointers,
 256                  * bypassing hash table). Currently algorithm scans them over
 257                  * and over again. Probably we should move busy objects out of
 258                  * LRU, or we can live with that.
 259                  */
 260                 if (nr-- == 0)
 261                         break;
 262                 if (atomic_read(&h->loh_ref) > 0)
 263                         continue;
 264                 hlist_del_init(&h->loh_hash);
 265                 list_move(&h->loh_lru, &dispose);
 266                 s->ls_total --;
 267         }
 268         write_unlock(&s->ls_guard);
 269         /*
 270          * Free everything on the dispose list. This is safe against races due
 271          * to the reasons described in lu_object_put().
 272          */
 273         while (!list_empty(&dispose)) {
 274                 h = container_of0(dispose.next,
 275                                  struct lu_object_header, loh_lru);
 276                 list_del_init(&h->loh_lru);
 277                 lu_object_free(env, lu_object_top(h));
 278                 s->ls_stats.s_lru_purged ++;
 279         }
 280         return nr;
 281 }
 282 EXPORT_SYMBOL(lu_site_purge);
 283
 284 /*
 285  * Object printing.
 286  *
 287  * Code below has to jump through certain loops to output object description
 288  * into libcfs_debug_msg-based log. The problem is that lu_object_print()
 289  * composes object description from strings that are parts of _lines_ of
 290  * output (i.e., strings that are not terminated by newline). This doesn't fit
 291  * very well into libcfs_debug_msg() interface that assumes that each message
 292  * supplied to it is a self-contained output line.
 293  *
 294  * To work around this, strings are collected in a temporary buffer
 295  * (implemented as a value of lu_cdebug_key key), until terminating newline
 296  * character is detected.
 297  *
 298  */
 299
 300 enum {
 301         /**
 302          * Maximal line size.
 303          *
 304          * XXX overflow is not handled correctly.
 305          */
 306         LU_CDEBUG_LINE = 256
 307 };
 308
 309 struct lu_cdebug_data {
 310         /**
 311          * Temporary buffer.
 312          */
 313         char lck_area[LU_CDEBUG_LINE];
 314 };
 315
 316 /* context key constructor/destructor: lu_global_key_init, lu_global_key_fini */
 317 LU_KEY_INIT_FINI(lu_global, struct lu_cdebug_data);
 318
 319 /**
 320  * Key, holding temporary buffer. This key is registered very early by
 321  * lu_global_init().
 322  */
 323 struct lu_context_key lu_global_key = {
 324         .lct_tags = LCT_MD_THREAD|LCT_DT_THREAD|LCT_CL_THREAD,
 325         .lct_init = lu_global_key_init,
 326         .lct_fini = lu_global_key_fini
 327 };
 328
 329 /**
 330  * Printer function emitting messages through libcfs_debug_msg().
 331  */
 332 int lu_cdebug_printer(const struct lu_env *env,
 333                       void *cookie, const char *format, ...)
 334 {
 335         struct lu_cdebug_print_info *info = cookie;
 336         struct lu_cdebug_data       *key;
 337         int used;
 338         int complete;
 339         va_list args;
 340
 341         va_start(args, format);
 342
 343         key = lu_context_key_get(&env->le_ctx, &lu_global_key);
 344         LASSERT(key != NULL);
 345
 346         used = strlen(key->lck_area);
 347         complete = format[strlen(format) - 1] == '\n';
 348         /*
 349          * Append new chunk to the buffer.
 350          */
 351         vsnprintf(key->lck_area + used,
 352                   ARRAY_SIZE(key->lck_area) - used, format, args);
 353         if (complete) {
 354                 libcfs_debug_msg(NULL, info->lpi_subsys, info->lpi_mask,
 355                                  (char *)info->lpi_file, info->lpi_fn,
 356                                  info->lpi_line, "%s", key->lck_area);
 357                 key->lck_area[0] = 0;
 358         }
 359         va_end(args);
 360         return 0;
 361 }
 362 EXPORT_SYMBOL(lu_cdebug_printer);
 363
 364 /*
 365  * Print object header.
 366  */
 367 static void lu_object_header_print(const struct lu_env *env,
 368                                    void *cookie, lu_printer_t printer,
 369                                    const struct lu_object_header *hdr)
 370 {
 371         (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
 372                    hdr, hdr->loh_flags, atomic_read(&hdr->loh_ref),
 373                    PFID(&hdr->loh_fid),
 374                    hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
 375                    list_empty(&hdr->loh_lru) ? "" : " lru",
 376                    hdr->loh_attr & LOHA_EXISTS ? " exist":"");
 377 }
 378
 379 /*
 380  * Print human readable representation of the @o to the @printer.
 381  */
 382 void lu_object_print(const struct lu_env *env, void *cookie,
 383                      lu_printer_t printer, const struct lu_object *o)
 384 {
 385         static const char ruler[] = "........................................";
 386         struct lu_object_header *top;
 387         int depth;
 388
 389         top = o->lo_header;
 390         lu_object_header_print(env, cookie, printer, top);
 391         (*printer)(env, cookie, "\n");
 392         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
 393                 depth = o->lo_depth + 4;
 394                 LASSERT(o->lo_ops->loo_object_print != NULL);
 395                 /*
 396                  * print `.' @depth times.
 397                  */
 398                 (*printer)(env, cookie, "%*.*s", depth, depth, ruler);
 399                 o->lo_ops->loo_object_print(env, cookie, printer, o);
 400                 (*printer)(env, cookie, "\n");
 401         }
 402 }
 403 EXPORT_SYMBOL(lu_object_print);
 404
 405 /*
 406  * Check object consistency.
 407  */
 408 int lu_object_invariant(const struct lu_object *o)
 409 {
 410         struct lu_object_header *top;
 411
 412         top = o->lo_header;
 413         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
 414                 if (o->lo_ops->loo_object_invariant != NULL &&
 415                     !o->lo_ops->loo_object_invariant(o))
 416                         return 0;
 417         }
 418         return 1;
 419 }
 420 EXPORT_SYMBOL(lu_object_invariant);
 421
 422 static struct lu_object *htable_lookup(struct lu_site *s,
 423                                        const struct hlist_head *bucket,
 424                                        const struct lu_fid *f)
 425 {
 426         struct lu_object_header *h;
 427         struct hlist_node *scan;
 428
 429         hlist_for_each_entry(h, scan, bucket, loh_hash) {
 430                 s->ls_stats.s_cache_check ++;
 431                 if (likely(lu_fid_eq(&h->loh_fid, f) &&
 432                            !lu_object_is_dying(h))) {
 433                         /* bump reference count... */
 434                         if (atomic_add_return(1, &h->loh_ref) == 1)
 435                                 ++ s->ls_busy;
 436                         /* and move to the head of the LRU */
 437                         /*
 438                          * XXX temporary disable this to measure effects of
 439                          * read-write locking.
 440                          */
 441                         /* list_move_tail(&h->loh_lru, &s->ls_lru); */
 442                         s->ls_stats.s_cache_hit ++;
 443                         return lu_object_top(h);
 444                 }
 445         }
 446         s->ls_stats.s_cache_miss ++;
 447         return NULL;
 448 }
 449
 450 static __u32 fid_hash(const struct lu_fid *f, int bits)
 451 {
 452         /* all objects with same id and different versions will belong to same
 453          * collisions list. */
 454         return hash_long(fid_flatten(f), bits);
 455 }
 456
 457 /*
 458  * Search cache for an object with the fid @f. If such object is found, return
 459  * it. Otherwise, create new object, insert it into cache and return it. In
 460  * any case, additional reference is acquired on the returned object.
 461  */
 462 struct lu_object *lu_object_find(const struct lu_env *env,
 463                                  struct lu_site *s, const struct lu_fid *f)
 464 {
 465         struct lu_object     *o;
 466         struct lu_object     *shadow;
 467         struct hlist_head *bucket;
 468
 469         /*
 470          * This uses standard index maintenance protocol:
 471          *
 472          *     - search index under lock, and return object if found;
 473          *     - otherwise, unlock index, allocate new object;
 474          *     - lock index and search again;
 475          *     - if nothing is found (usual case), insert newly created
 476          *       object into index;
 477          *     - otherwise (race: other thread inserted object), free
 478          *       object just allocated.
 479          *     - unlock index;
 480          *     - return object.
 481          */
 482
 483         bucket = s->ls_hash + fid_hash(f, s->ls_hash_bits);
 484
 485         read_lock(&s->ls_guard);
 486         o = htable_lookup(s, bucket, f);
 487         read_unlock(&s->ls_guard);
 488
 489         if (o != NULL)
 490                 return o;
 491
 492         /*
 493          * Allocate new object. This may result in rather complicated
 494          * operations, including fld queries, inode loading, etc.
 495          */
 496         o = lu_object_alloc(env, s, f);
 497         if (unlikely(IS_ERR(o)))
 498                 return o;
 499
 500         LASSERT(lu_fid_eq(lu_object_fid(o), f));
 501
 502         write_lock(&s->ls_guard);
 503         shadow = htable_lookup(s, bucket, f);
 504         if (likely(shadow == NULL)) {
 505                 hlist_add_head(&o->lo_header->loh_hash, bucket);
 506                 list_add_tail(&o->lo_header->loh_lru, &s->ls_lru);
 507                 ++ s->ls_busy;
 508                 ++ s->ls_total;
 509                 shadow = o;
 510                 o = NULL;
 511         } else
 512                 s->ls_stats.s_cache_race ++;
 513         write_unlock(&s->ls_guard);
 514         if (o != NULL)
 515                 lu_object_free(env, o);
 516         return shadow;
 517 }
 518 EXPORT_SYMBOL(lu_object_find);
 519
 520 /*
 521 /**
 522  * Global list of all device types.
 523  */
 524 static CFS_LIST_HEAD(lu_device_types);
 525
 526 int lu_device_type_init(struct lu_device_type *ldt)
 527 {
 528         int result;
 529
 530         CFS_INIT_LIST_HEAD(&ldt->ldt_linkage);
 531         result = ldt->ldt_ops->ldto_init(ldt);
 532         if (result == 0)
 533                 list_add(&ldt->ldt_linkage, &lu_device_types);
 534         return result;
 535 }
 536 EXPORT_SYMBOL(lu_device_type_init);
 537
 538 void lu_device_type_fini(struct lu_device_type *ldt)
 539 {
 540         list_del_init(&ldt->ldt_linkage);
 541         ldt->ldt_ops->ldto_fini(ldt);
 542 }
 543 EXPORT_SYMBOL(lu_device_type_fini);
 544
 545 void lu_types_stop(void)
 546 {
 547         struct lu_device_type *ldt;
 548
 549         list_for_each_entry(ldt, &lu_device_types, ldt_linkage) {
 550                 if (ldt->ldt_device_nr == 0)
 551                         ldt->ldt_ops->ldto_stop(ldt);
 552         }
 553 }
 554 EXPORT_SYMBOL(lu_types_stop);
 555
 556 /**
 557  * Global list of all sites on this node
 558  */
 559 static CFS_LIST_HEAD(lu_sites);
 560 static DECLARE_MUTEX(lu_sites_guard);
 561
 562 /**
 563  * Global environment used by site shrinker.
 564  */
 565 static struct lu_env lu_shrink_env;
 566
 567 /**
 568  * Print all objects in \a s.
 569  */
 570 void lu_site_print(const struct lu_env *env, struct lu_site *s, void *cookie,
 571                    lu_printer_t printer)
 572 {
 573         int i;
 574
 575         for (i = 0; i < s->ls_hash_size; ++i) {
 576                 struct lu_object_header *h;
 577                 struct hlist_node       *scan;
 578
 579                 read_lock(&s->ls_guard);
 580                 hlist_for_each_entry(h, scan, &s->ls_hash[i], loh_hash) {
 581
 582                         if (!list_empty(&h->loh_layers)) {
 583                                 const struct lu_object *obj;
 584
 585                                 obj = lu_object_top(h);
 586                                 lu_object_print(env, cookie, printer, obj);
 587                         } else
 588                                 lu_object_header_print(env, cookie, printer, h);
 589                 }
 590                 read_unlock(&s->ls_guard);
 591         }
 592 }
 593 EXPORT_SYMBOL(lu_site_print);
 594
 595 enum {
 596         LU_CACHE_PERCENT   = 20,
 597 };
 598
 599 /**
 600  * Return desired hash table order.
 601  */
 602 static int lu_htable_order(void)
 603 {
 604         unsigned long cache_size;
 605         int bits;
 606
 607         /*
 608          * Calculate hash table size, assuming that we want reasonable
 609          * performance when 20% of total memory is occupied by cache of
 610          * lu_objects.
 611          *
 612          * Size of lu_object is (arbitrary) taken as 1K (together with inode).
 613          */
 614         cache_size = num_physpages;
 615
 616 #if BITS_PER_LONG == 32
 617         /* limit hashtable size for lowmem systems to low RAM */
 618         if (cache_size > 1 << (30 - CFS_PAGE_SHIFT))
 619                 cache_size = 1 << (30 - CFS_PAGE_SHIFT) * 3 / 4;
 620 #endif
 621
 622         cache_size = cache_size / 100 * LU_CACHE_PERCENT *
 623                 (CFS_PAGE_SIZE / 1024);
 624
 625         for (bits = 1; (1 << bits) < cache_size; ++bits) {
 626                 ;
 627         }
 628         return bits;
 629 }
 630
 631 static struct lock_class_key lu_site_guard_class;
 632
 633 /**
 634  * Initialize site \a s, with \a d as the top level device.
 635  */
 636 int lu_site_init(struct lu_site *s, struct lu_device *top)
 637 {
 638         int bits;
 639         int size;
 640         int i;
 641         ENTRY;
 642
 643         memset(s, 0, sizeof *s);
 644         rwlock_init(&s->ls_guard);
 645         lockdep_set_class(&s->ls_guard, &lu_site_guard_class);
 646         CFS_INIT_LIST_HEAD(&s->ls_lru);
 647         CFS_INIT_LIST_HEAD(&s->ls_linkage);
 648         cfs_waitq_init(&s->ls_marche_funebre);
 649         s->ls_top_dev = top;
 650         top->ld_site = s;
 651         lu_device_get(top);
 652         lu_ref_add(&top->ld_reference, "site-top", s);
 653
 654         for (bits = lu_htable_order(), size = 1 << bits;
 655              (s->ls_hash =
 656               cfs_alloc_large(size * sizeof s->ls_hash[0])) == NULL;
 657              --bits, size >>= 1) {
 658                 /*
 659                  * Scale hash table down, until allocation succeeds.
 660                  */
 661                 ;
 662         }
 663
 664         s->ls_hash_size = size;
 665         s->ls_hash_bits = bits;
 666         s->ls_hash_mask = size - 1;
 667
 668         for (i = 0; i < size; i++)
 669                 INIT_HLIST_HEAD(&s->ls_hash[i]);
 670
 671         RETURN(0);
 672 }
 673 EXPORT_SYMBOL(lu_site_init);
 674
 675 /**
 676  * Finalize \a s and release its resources.
 677  */
 678 void lu_site_fini(struct lu_site *s)
 679 {
 680         LASSERT(list_empty(&s->ls_lru));
 681         LASSERT(s->ls_total == 0);
 682
 683         down(&lu_sites_guard);
 684         list_del_init(&s->ls_linkage);
 685         up(&lu_sites_guard);
 686
 687         if (s->ls_hash != NULL) {
 688                 int i;
 689                 for (i = 0; i < s->ls_hash_size; i++)
 690                         LASSERT(hlist_empty(&s->ls_hash[i]));
 691                 cfs_free_large(s->ls_hash);
 692                 s->ls_hash = NULL;
 693         }
 694         if (s->ls_top_dev != NULL) {
 695                 s->ls_top_dev->ld_site = NULL;
 696                 lu_ref_del(&s->ls_top_dev->ld_reference, "site-top", s);
 697                 lu_device_put(s->ls_top_dev);
 698                 s->ls_top_dev = NULL;
 699         }
 700 }
 701 EXPORT_SYMBOL(lu_site_fini);
 702
 703 /**
 704  * Called when initialization of stack for this site is completed.
 705  */
 706 int lu_site_init_finish(struct lu_site *s)
 707 {
 708         int result;
 709         down(&lu_sites_guard);
 710         result = lu_context_refill(&lu_shrink_env.le_ctx);
 711         if (result == 0)
 712                 list_add(&s->ls_linkage, &lu_sites);
 713         up(&lu_sites_guard);
 714         return result;
 715 }
 716 EXPORT_SYMBOL(lu_site_init_finish);
 717
 718 /**
 719  * Acquire additional reference on device \a d
 720  */
 721 void lu_device_get(struct lu_device *d)
 722 {
 723         atomic_inc(&d->ld_ref);
 724 }
 725 EXPORT_SYMBOL(lu_device_get);
 726
 727 /**
 728  * Release reference on device \a d.
 729  */
 730 void lu_device_put(struct lu_device *d)
 731 {
 732         LASSERT(atomic_read(&d->ld_ref) > 0);
 733         atomic_dec(&d->ld_ref);
 734 }
 735 EXPORT_SYMBOL(lu_device_put);
 736
 737 /**
 738  * Initialize device \a d of type \a t.
 739  */
 740 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
 741 {
 742         if (t->ldt_device_nr++ == 0 && t->ldt_ops->ldto_start != NULL)
 743                 t->ldt_ops->ldto_start(t);
 744         memset(d, 0, sizeof *d);
 745         atomic_set(&d->ld_ref, 0);
 746         d->ld_type = t;
 747         lu_ref_init(&d->ld_reference);
 748         return 0;
 749 }
 750 EXPORT_SYMBOL(lu_device_init);
 751
 752 /**
 753  * Finalize device \a d.
 754  */
 755 void lu_device_fini(struct lu_device *d)
 756 {
 757         struct lu_device_type *t;
 758
 759         t = d->ld_type;
 760         if (d->ld_obd != NULL)
 761                 /* finish lprocfs */
 762                 lprocfs_obd_cleanup(d->ld_obd);
 763
 764         lu_ref_fini(&d->ld_reference);
 765         LASSERTF(atomic_read(&d->ld_ref) == 0,
 766                  "Refcount is %u\n", atomic_read(&d->ld_ref));
 767         LASSERT(t->ldt_device_nr > 0);
 768         if (--t->ldt_device_nr == 0 && t->ldt_ops->ldto_stop != NULL)
 769                 t->ldt_ops->ldto_stop(t);
 770 }
 771 EXPORT_SYMBOL(lu_device_fini);
 772
 773 /**
 774  * Initialize object \a o that is part of compound object \a h and was created
 775  * by device \a d.
 776  */
 777 int lu_object_init(struct lu_object *o,
 778                    struct lu_object_header *h, struct lu_device *d)
 779 {
 780         memset(o, 0, sizeof *o);
 781         o->lo_header = h;
 782         o->lo_dev    = d;
 783         lu_device_get(d);
 784         o->lo_dev_ref = lu_ref_add(&d->ld_reference, "lu_object", o);
 785         CFS_INIT_LIST_HEAD(&o->lo_linkage);
 786         return 0;
 787 }
 788 EXPORT_SYMBOL(lu_object_init);
 789
 790 /**
 791  * Finalize object and release its resources.
 792  */
 793 void lu_object_fini(struct lu_object *o)
 794 {
 795         struct lu_device *dev = o->lo_dev;
 796
 797         LASSERT(list_empty(&o->lo_linkage));
 798
 799         if (dev != NULL) {
 800                 lu_ref_del_at(&dev->ld_reference,
 801                               o->lo_dev_ref , "lu_object", o);
 802                 lu_device_put(dev);
 803                 o->lo_dev = NULL;
 804         }
 805 }
 806 EXPORT_SYMBOL(lu_object_fini);
 807
 808 /**
 809  * Add object \a o as first layer of compound object \a h
 810  *
 811  * This is typically called by the ->ldo_object_alloc() method of top-level
 812  * device.
 813  */
 814 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
 815 {
 816         list_move(&o->lo_linkage, &h->loh_layers);
 817 }
 818 EXPORT_SYMBOL(lu_object_add_top);
 819
 820 /**
 821  * Add object \a o as a layer of compound object, going after \a before.
 822  *
 823  * This is typically called by the ->ldo_object_alloc() method of \a
 824  * before->lo_dev.
 825  */
 826 void lu_object_add(struct lu_object *before, struct lu_object *o)
 827 {
 828         list_move(&o->lo_linkage, &before->lo_linkage);
 829 }
 830 EXPORT_SYMBOL(lu_object_add);
 831
 832 /**
 833  * Initialize compound object.
 834  */
 835 int lu_object_header_init(struct lu_object_header *h)
 836 {
 837         memset(h, 0, sizeof *h);
 838         atomic_set(&h->loh_ref, 1);
 839         INIT_HLIST_NODE(&h->loh_hash);
 840         CFS_INIT_LIST_HEAD(&h->loh_lru);
 841         CFS_INIT_LIST_HEAD(&h->loh_layers);
 842         lu_ref_init(&h->loh_reference);
 843         return 0;
 844 }
 845 EXPORT_SYMBOL(lu_object_header_init);
 846
 847 /**
 848  * Finalize compound object.
 849  */
 850 void lu_object_header_fini(struct lu_object_header *h)
 851 {
 852         LASSERT(list_empty(&h->loh_layers));
 853         LASSERT(list_empty(&h->loh_lru));
 854         LASSERT(hlist_unhashed(&h->loh_hash));
 855         lu_ref_fini(&h->loh_reference);
 856 }
 857 EXPORT_SYMBOL(lu_object_header_fini);
 858
 859 /**
 860  * Given a compound object, find its slice, corresponding to the device type
 861  * \a dtype.
 862  */
 863 struct lu_object *lu_object_locate(struct lu_object_header *h,
 864                                    const struct lu_device_type *dtype)
 865 {
 866         struct lu_object *o;
 867
 868         list_for_each_entry(o, &h->loh_layers, lo_linkage) {
 869                 if (o->lo_dev->ld_type == dtype)
 870                         return o;
 871         }
 872         return NULL;
 873 }
 874 EXPORT_SYMBOL(lu_object_locate);
 875
 876
 877
 878 /**
 879  * Finalize and free devices in the device stack.
 880  *
 881  * Finalize device stack by purging object cache, and calling
 882  * lu_device_type_operations::ldto_device_fini() and
 883  * lu_device_type_operations::ldto_device_free() on all devices in the stack.
 884  */
 885 void lu_stack_fini(const struct lu_env *env, struct lu_device *top)
 886 {
 887         struct lu_site   *site = top->ld_site;
 888         struct lu_device *scan;
 889         struct lu_device *next;
 890
 891         lu_site_purge(env, site, ~0);
 892         for (scan = top; scan != NULL; scan = next) {
 893                 next = scan->ld_type->ldt_ops->ldto_device_fini(env, scan);
 894                 lu_ref_del(&scan->ld_reference, "lu-stack", &lu_site_init);
 895                 lu_device_put(scan);
 896         }
 897
 898         /* purge again. */
 899         lu_site_purge(env, site, ~0);
 900
 901         if (!list_empty(&site->ls_lru) || site->ls_total != 0) {
 902                 /*
 903                  * Uh-oh, objects still exist.
 904                  */
 905                 static DECLARE_LU_CDEBUG_PRINT_INFO(cookie, D_ERROR);
 906
 907                 lu_site_print(env, site, &cookie, lu_cdebug_printer);
 908         }
 909
 910         for (scan = top; scan != NULL; scan = next) {
 911                 const struct lu_device_type *ldt = scan->ld_type;
 912                 struct obd_type             *type;
 913
 914                 next = ldt->ldt_ops->ldto_device_free(env, scan);
 915                 type = ldt->ldt_obd_type;
 916                 if (type != NULL) {
 917                 type->typ_refcnt--;
 918                 class_put_type(type);
 919         }
 920         }
 921 }
 922 EXPORT_SYMBOL(lu_stack_fini);
 923
 924 enum {
 925         /**
 926          * Maximal number of tld slots.
 927          */
 928         LU_CONTEXT_KEY_NR = 32
 929 };
 930
 931 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
 932
 933 static spinlock_t lu_keys_guard = SPIN_LOCK_UNLOCKED;
 934
 935 /**
 936  * Global counter incremented whenever key is registered, unregistered,
 937  * revived or quiesced. This is used to void unnecessary calls to
 938  * lu_context_refill(). No locking is provided, as initialization and shutdown
 939  * are supposed to be externally serialized.
 940  */
 941 static unsigned key_set_version = 0;
 942
 943 /**
 944  * Register new key.
 945  */
 946 int lu_context_key_register(struct lu_context_key *key)
 947 {
 948         int result;
 949         int i;
 950
 951         LASSERT(key->lct_init != NULL);
 952         LASSERT(key->lct_fini != NULL);
 953         LASSERT(key->lct_tags != 0);
 954         LASSERT(key->lct_owner != NULL);
 955
 956         result = -ENFILE;
 957         spin_lock(&lu_keys_guard);
 958         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
 959                 if (lu_keys[i] == NULL) {
 960                         key->lct_index = i;
 961                         atomic_set(&key->lct_used, 1);
 962                         lu_keys[i] = key;
 963                         lu_ref_init(&key->lct_reference);
 964                         result = 0;
 965                         ++key_set_version;
 966                         break;
 967                 }
 968         }
 969         spin_unlock(&lu_keys_guard);
 970         return result;
 971 }
 972 EXPORT_SYMBOL(lu_context_key_register);
 973
 974 static void key_fini(struct lu_context *ctx, int index)
 975 {
 976         if (ctx->lc_value[index] != NULL) {
 977                 struct lu_context_key *key;
 978
 979                 key = lu_keys[index];
 980                 LASSERT(key != NULL);
 981                 LASSERT(key->lct_fini != NULL);
 982                 LASSERT(atomic_read(&key->lct_used) > 1);
 983
 984                 key->lct_fini(ctx, key, ctx->lc_value[index]);
 985                 lu_ref_del(&key->lct_reference, "ctx", ctx);
 986                 atomic_dec(&key->lct_used);
 987                 LASSERT(key->lct_owner != NULL);
 988                 if (!(ctx->lc_tags & LCT_NOREF)) {
 989                         LASSERT(module_refcount(key->lct_owner) > 0);
 990                         module_put(key->lct_owner);
 991                 }
 992                 ctx->lc_value[index] = NULL;
 993         }
 994 }
 995
 996 /**
 997  * Deregister key.
 998  */
 999 void lu_context_key_degister(struct lu_context_key *key)
1000 {
1001         LASSERT(atomic_read(&key->lct_used) >= 1);
1002         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1003
1004         ++key_set_version;
1005         key_fini(&lu_shrink_env.le_ctx, key->lct_index);
1006
1007         if (atomic_read(&key->lct_used) > 1)
1008                 CERROR("key has instances.\n");
1009         spin_lock(&lu_keys_guard);
1010         lu_keys[key->lct_index] = NULL;
1011         spin_unlock(&lu_keys_guard);
1012 }
1013 EXPORT_SYMBOL(lu_context_key_degister);
1014
1015 /**
1016  * Register a number of keys. This has to be called after all keys have been
1017  * initialized by a call to LU_CONTEXT_KEY_INIT().
1018  */
1019 int lu_context_key_register_many(struct lu_context_key *k, ...)
1020 {
1021         struct lu_context_key *key = k;
1022         va_list args;
1023         int result;
1024
1025         va_start(args, k);
1026         do {
1027                 result = lu_context_key_register(key);
1028                 if (result)
1029                         break;
1030                 key = va_arg(args, struct lu_context_key *);
1031         } while (key != NULL);
1032         va_end(args);
1033
1034         if (result != 0) {
1035                 va_start(args, k);
1036                 while (k != key) {
1037                         lu_context_key_degister(k);
1038                         k = va_arg(args, struct lu_context_key *);
1039                 }
1040                 va_end(args);
1041         }
1042
1043         return result;
1044 }
1045 EXPORT_SYMBOL(lu_context_key_register_many);
1046
1047 /**
1048  * De-register a number of keys. This is a dual to
1049  * lu_context_key_register_many().
1050  */
1051 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1052 {
1053         va_list args;
1054
1055         va_start(args, k);
1056         do {
1057                 lu_context_key_degister(k);
1058                 k = va_arg(args, struct lu_context_key*);
1059         } while (k != NULL);
1060         va_end(args);
1061 }
1062 EXPORT_SYMBOL(lu_context_key_degister_many);
1063
1064 /**
1065  * Revive a number of keys.
1066  */
1067 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1068 {
1069         va_list args;
1070
1071         va_start(args, k);
1072         do {
1073                 lu_context_key_revive(k);
1074                 k = va_arg(args, struct lu_context_key*);
1075         } while (k != NULL);
1076         va_end(args);
1077 }
1078 EXPORT_SYMBOL(lu_context_key_revive_many);
1079
1080 /**
1081  * Quiescent a number of keys.
1082  */
1083 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1084 {
1085         va_list args;
1086
1087         va_start(args, k);
1088         do {
1089                 lu_context_key_quiesce(k);
1090                 k = va_arg(args, struct lu_context_key*);
1091         } while (k != NULL);
1092         va_end(args);
1093 }
1094 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1095
1096 /**
1097  * Return value associated with key \a key in context \a ctx.
1098  */
1099 void *lu_context_key_get(const struct lu_context *ctx,
1100                          const struct lu_context_key *key)
1101 {
1102         LINVRNT(ctx->lc_state == LCS_ENTERED);
1103         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1104         return ctx->lc_value[key->lct_index];
1105 }
1106 EXPORT_SYMBOL(lu_context_key_get);
1107
1108 /**
1109  * List of remembered contexts. XXX document me.
1110  */
1111 static CFS_LIST_HEAD(lu_context_remembered);
1112
1113 /**
1114  * Destroy \a key in all remembered contexts. This is used to destroy key
1115  * values in "shared" contexts (like service threads), when a module owning
1116  * the key is about to be unloaded.
1117  */
1118 void lu_context_key_quiesce(struct lu_context_key *key)
1119 {
1120         struct lu_context *ctx;
1121
1122         if (!(key->lct_tags & LCT_QUIESCENT)) {
1123                 key->lct_tags |= LCT_QUIESCENT;
1124                 /*
1125                  * XXX memory barrier has to go here.
1126                  */
1127                 spin_lock(&lu_keys_guard);
1128                 list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
1129                         key_fini(ctx, key->lct_index);
1130                 spin_unlock(&lu_keys_guard);
1131                 ++key_set_version;
1132         }
1133 }
1134 EXPORT_SYMBOL(lu_context_key_quiesce);
1135
1136 void lu_context_key_revive(struct lu_context_key *key)
1137 {
1138         key->lct_tags &= ~LCT_QUIESCENT;
1139         ++key_set_version;
1140 }
1141 EXPORT_SYMBOL(lu_context_key_revive);
1142
1143 static void keys_fini(struct lu_context *ctx)
1144 {
1145         int i;
1146
1147         if (ctx->lc_value != NULL) {
1148                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1149                         key_fini(ctx, i);
1150                 OBD_FREE(ctx->lc_value,
1151                          ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1152                 ctx->lc_value = NULL;
1153         }
1154 }
1155
1156 static int keys_fill(struct lu_context *ctx)
1157 {
1158         int i;
1159
1160         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1161                 struct lu_context_key *key;
1162
1163                 key = lu_keys[i];
1164                 if (ctx->lc_value[i] == NULL && key != NULL &&
1165                     (key->lct_tags & ctx->lc_tags) &&
1166                     /*
1167                      * Don't create values for a LCT_QUIESCENT key, as this
1168                      * will pin module owning a key.
1169                      */
1170                     !(key->lct_tags & LCT_QUIESCENT)) {
1171                         void *value;
1172
1173                         LINVRNT(key->lct_init != NULL);
1174                         LINVRNT(key->lct_index == i);
1175
1176                         value = key->lct_init(ctx, key);
1177                         if (unlikely(IS_ERR(value)))
1178                                 return PTR_ERR(value);
1179                         LASSERT(key->lct_owner != NULL);
1180                         if (!(ctx->lc_tags & LCT_NOREF))
1181                                 try_module_get(key->lct_owner);
1182                         lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1183                         atomic_inc(&key->lct_used);
1184                         /*
1185                          * This is the only place in the code, where an
1186                          * element of ctx->lc_value[] array is set to non-NULL
1187                          * value.
1188                          */
1189                         ctx->lc_value[i] = value;
1190                         if (key->lct_exit != NULL)
1191                                 ctx->lc_tags |= LCT_HAS_EXIT;
1192                 }
1193                 ctx->lc_version = key_set_version;
1194         }
1195         return 0;
1196 }
1197
1198 static int keys_init(struct lu_context *ctx)
1199 {
1200         int result;
1201
1202         OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1203         if (likely(ctx->lc_value != NULL))
1204                 result = keys_fill(ctx);
1205         else
1206                 result = -ENOMEM;
1207
1208         if (result != 0)
1209                 keys_fini(ctx);
1210         return result;
1211 }
1212
1213 /**
1214  * Initialize context data-structure. Create values for all keys.
1215  */
1216 int lu_context_init(struct lu_context *ctx, __u32 tags)
1217 {
1218         memset(ctx, 0, sizeof *ctx);
1219         ctx->lc_state = LCS_INITIALIZED;
1220         ctx->lc_tags = tags;
1221         if (tags & LCT_REMEMBER) {
1222                 spin_lock(&lu_keys_guard);
1223                 list_add(&ctx->lc_remember, &lu_context_remembered);
1224                 spin_unlock(&lu_keys_guard);
1225         } else
1226                 CFS_INIT_LIST_HEAD(&ctx->lc_remember);
1227         return keys_init(ctx);
1228 }
1229 EXPORT_SYMBOL(lu_context_init);
1230
1231 /**
1232  * Finalize context data-structure. Destroy key values.
1233  */
1234 void lu_context_fini(struct lu_context *ctx)
1235 {
1236         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1237         ctx->lc_state = LCS_FINALIZED;
1238         keys_fini(ctx);
1239         spin_lock(&lu_keys_guard);
1240         list_del_init(&ctx->lc_remember);
1241         spin_unlock(&lu_keys_guard);
1242 }
1243 EXPORT_SYMBOL(lu_context_fini);
1244
1245 /**
1246  * Called before entering context.
1247  */
1248 void lu_context_enter(struct lu_context *ctx)
1249 {
1250         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1251         ctx->lc_state = LCS_ENTERED;
1252 }
1253 EXPORT_SYMBOL(lu_context_enter);
1254
1255 /**
1256  * Called after exiting from \a ctx
1257  */
1258 void lu_context_exit(struct lu_context *ctx)
1259 {
1260         int i;
1261
1262         LINVRNT(ctx->lc_state == LCS_ENTERED);
1263         ctx->lc_state = LCS_LEFT;
1264         if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
1265                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1266                         if (ctx->lc_value[i] != NULL) {
1267                                 struct lu_context_key *key;
1268
1269                                 key = lu_keys[i];
1270                                 LASSERT(key != NULL);
1271                                 if (key->lct_exit != NULL)
1272                                         key->lct_exit(ctx,
1273                                                       key, ctx->lc_value[i]);
1274                         }
1275                 }
1276         }
1277 }
1278 EXPORT_SYMBOL(lu_context_exit);
1279
1280 /**
1281  * Allocate for context all missing keys that were registered after context
1282  * creation.
1283  */
1284 int lu_context_refill(struct lu_context *ctx)
1285 {
1286         LINVRNT(ctx->lc_value != NULL);
1287         return ctx->lc_version == key_set_version ? 0 : keys_fill(ctx);
1288 }
1289 EXPORT_SYMBOL(lu_context_refill);
1290
1291 static int lu_env_setup(struct lu_env *env, struct lu_context *ses,
1292                         __u32 tags, int noref)
1293 {
1294         int result;
1295
1296         LINVRNT(ergo(!noref, !(tags & LCT_NOREF)));
1297
1298         env->le_ses = ses;
1299         result = lu_context_init(&env->le_ctx, tags);
1300         if (likely(result == 0))
1301                 lu_context_enter(&env->le_ctx);
1302         return result;
1303 }
1304
1305 static int lu_env_init_noref(struct lu_env *env, struct lu_context *ses,
1306                              __u32 tags)
1307 {
1308         return lu_env_setup(env, ses, tags, 1);
1309 }
1310
1311 int lu_env_init(struct lu_env *env, struct lu_context *ses, __u32 tags)
1312 {
1313         return lu_env_setup(env, ses, tags, 0);
1314 }
1315 EXPORT_SYMBOL(lu_env_init);
1316
1317 void lu_env_fini(struct lu_env *env)
1318 {
1319         lu_context_exit(&env->le_ctx);
1320         lu_context_fini(&env->le_ctx);
1321         env->le_ses = NULL;
1322 }
1323 EXPORT_SYMBOL(lu_env_fini);
1324
1325 int lu_env_refill(struct lu_env *env)
1326 {
1327         int result;
1328
1329         result = lu_context_refill(&env->le_ctx);
1330         if (result == 0 && env->le_ses != NULL)
1331                 result = lu_context_refill(env->le_ses);
1332         return result;
1333 }
1334 EXPORT_SYMBOL(lu_env_refill);
1335
1336 static struct shrinker *lu_site_shrinker = NULL;
1337
1338 #ifdef __KERNEL__
1339 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1340 {
1341         struct lu_site *s;
1342         struct lu_site *tmp;
1343         int cached = 0;
1344         int remain = nr;
1345         CFS_LIST_HEAD(splice);
1346
1347         if (nr != 0 && !(gfp_mask & __GFP_FS))
1348                 return -1;
1349
1350         down(&lu_sites_guard);
1351         list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1352                 if (nr != 0) {
1353                         remain = lu_site_purge(&lu_shrink_env, s, remain);
1354                         /*
1355                          * Move just shrunk site to the tail of site list to
1356                          * assure shrinking fairness.
1357                          */
1358                         list_move_tail(&s->ls_linkage, &splice);
1359                 }
1360                 read_lock(&s->ls_guard);
1361                 cached += s->ls_total - s->ls_busy;
1362                 read_unlock(&s->ls_guard);
1363                 if (remain <= 0)
1364                         break;
1365         }
1366         list_splice(&splice, lu_sites.prev);
1367         up(&lu_sites_guard);
1368         return cached;
1369 }
1370
1371 #else  /* !__KERNEL__ */
1372 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1373 {
1374         return 0;
1375 }
1376 #endif /* __KERNEL__ */
1377
1378 int  lu_ref_global_init(void);
1379 void lu_ref_global_fini(void);
1380
1381 /**
1382  * Initialization of global lu_* data.
1383  */
1384 int lu_global_init(void)
1385 {
1386         int result;
1387
1388         CDEBUG(D_CONSOLE, "Lustre LU module (%p).\n", &lu_keys);
1389
1390         LU_CONTEXT_KEY_INIT(&lu_global_key);
1391         result = lu_context_key_register(&lu_global_key);
1392         if (result != 0)
1393                 return result;
1394                 /*
1395          * At this level, we don't know what tags are needed, so allocate them
1396          * conservatively. This should not be too bad, because this
1397          * environment is global.
1398                  */
1399                 down(&lu_sites_guard);
1400                 result = lu_env_init_noref(&lu_shrink_env, NULL, LCT_SHRINKER);
1401                 up(&lu_sites_guard);
1402         if (result != 0)
1403                 return result;
1404
1405         result = lu_ref_global_init();
1406         if (result != 0)
1407                 return result;
1408                         /*
1409          * seeks estimation: 3 seeks to read a record from oi, one to read
1410          * inode, one for ea. Unfortunately setting this high value results in
1411          * lu_object/inode cache consuming all the memory.
1412          */
1413         lu_site_shrinker = set_shrinker(DEFAULT_SEEKS, lu_cache_shrink);
1414         if (lu_site_shrinker == NULL)
1415                 return -ENOMEM;
1416
1417                                 result = lu_time_global_init();
1418         return result;
1419 }
1420
1421 /**
1422  * Dual to lu_global_init().
1423  */
1424 void lu_global_fini(void)
1425 {
1426         lu_time_global_fini();
1427         if (lu_site_shrinker != NULL) {
1428                 remove_shrinker(lu_site_shrinker);
1429                 lu_site_shrinker = NULL;
1430         }
1431
1432         lu_context_key_degister(&lu_global_key);
1433
1434         /*
1435          * Tear shrinker environment down _after_ de-registering
1436          * lu_global_key, because the latter has a value in the former.
1437          */
1438         down(&lu_sites_guard);
1439         lu_env_fini(&lu_shrink_env);
1440         up(&lu_sites_guard);
1441
1442         lu_ref_global_fini();
1443 }
1444
1445 struct lu_buf LU_BUF_NULL = {
1446         .lb_buf = NULL,
1447         .lb_len = 0
1448 };
1449 EXPORT_SYMBOL(LU_BUF_NULL);
1450
1451 /*
1452  * XXX: Functions below logically belong to fid module, but they are used by
1453  * dt_store_open(). Put them here until better place is found.
1454  */
1455
1456 void fid_pack(struct lu_fid_pack *pack, const struct lu_fid *fid,
1457               struct lu_fid *befider)
1458 {
1459         int recsize;
1460         __u64 seq;
1461         __u32 oid;
1462
1463         seq = fid_seq(fid);
1464         oid = fid_oid(fid);
1465
1466         /*
1467          * Two cases: compact 6 bytes representation for a common case, and
1468          * full 17 byte representation for "unusual" fid.
1469          */
1470
1471         /*
1472          * Check that usual case is really usual.
1473          */
1474         CLASSERT(LUSTRE_SEQ_MAX_WIDTH < 0xffffull);
1475
1476         if (fid_is_igif(fid) ||
1477             seq > 0xffffffull || oid > 0xffff || fid_ver(fid) != 0) {
1478                 fid_cpu_to_be(befider, fid);
1479                 recsize = sizeof *befider;
1480         } else {
1481                 unsigned char *small_befider;
1482
1483                 small_befider = (char *)befider;
1484
1485                 small_befider[0] = seq >> 16;
1486                 small_befider[1] = seq >> 8;
1487                 small_befider[2] = seq;
1488
1489                 small_befider[3] = oid >> 8;
1490                 small_befider[4] = oid;
1491
1492                 recsize = 5;
1493         }
1494         memcpy(pack->fp_area, befider, recsize);
1495         pack->fp_len = recsize + 1;
1496 }
1497 EXPORT_SYMBOL(fid_pack);
1498
1499 int fid_unpack(const struct lu_fid_pack *pack, struct lu_fid *fid)
1500 {
1501         int result;
1502
1503         result = 0;
1504         switch (pack->fp_len) {
1505         case sizeof *fid + 1:
1506                 memcpy(fid, pack->fp_area, sizeof *fid);
1507                 fid_be_to_cpu(fid, fid);
1508                 break;
1509         case 6: {
1510                 const unsigned char *area;
1511
1512                 area = pack->fp_area;
1513                 fid->f_seq = (area[0] << 16) | (area[1] << 8) | area[2];
1514                 fid->f_oid = (area[3] << 8) | area[4];
1515                 fid->f_ver = 0;
1516                 break;
1517         }
1518         default:
1519                 CERROR("Unexpected packed fid size: %d\n", pack->fp_len);
1520                 result = -EIO;
1521         }
1522         return result;
1523 }
1524 EXPORT_SYMBOL(fid_unpack);
1525
1526 const char *lu_time_names[LU_TIME_NR] = {
1527         [LU_TIME_FIND_LOOKUP] = "find_lookup",
1528         [LU_TIME_FIND_ALLOC]  = "find_alloc",
1529         [LU_TIME_FIND_INSERT] = "find_insert"
1530 };
1531 EXPORT_SYMBOL(lu_time_names);
1532
1533 /**
1534  * Helper function to initialize a number of kmem slab caches at once.
1535  */
1536 int lu_kmem_init(struct lu_kmem_descr *caches)
1537 {
1538         int result;
1539
1540         for (result = 0; caches->ckd_cache != NULL; ++caches) {
1541                 *caches->ckd_cache = cfs_mem_cache_create(caches->ckd_name,
1542                                                           caches->ckd_size,
1543                                                           0, 0);
1544                 if (*caches->ckd_cache == NULL) {
1545                         result = -ENOMEM;
1546                         break;
1547                 }
1548         }
1549         return result;
1550 }
1551 EXPORT_SYMBOL(lu_kmem_init);
1552
1553 /**
1554  * Helper function to finalize a number of kmem slab cached at once. Dual to
1555  * lu_kmem_init().
1556  */
1557 void lu_kmem_fini(struct lu_kmem_descr *caches)
1558 {
1559         int rc;
1560
1561         for (; caches->ckd_cache != NULL; ++caches) {
1562                 if (*caches->ckd_cache != NULL) {
1563                         rc = cfs_mem_cache_destroy(*caches->ckd_cache);
1564                         LASSERTF(rc == 0, "couldn't destroy %s slab\n",
1565                                  caches->ckd_name);
1566                         *caches->ckd_cache = NULL;
1567                 }
1568         }
1569 }
1570 EXPORT_SYMBOL(lu_kmem_fini);
1571