Whamcloud - gitweb
4aa04430eede20f9b12c20917c061972498a2ed9
[fs/lustre-release.git] / 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                 if (cdebug_show(info->lpi_mask, info->lpi_subsys))
355                         libcfs_debug_msg(NULL, info->lpi_subsys, info->lpi_mask,
356                                          (char *)info->lpi_file, info->lpi_fn,
357                                          info->lpi_line, "%s", key->lck_area);
358                 key->lck_area[0] = 0;
359         }
360         va_end(args);
361         return 0;
362 }
363 EXPORT_SYMBOL(lu_cdebug_printer);
364
365 /**
366  * Print object header.
367  */
368 void lu_object_header_print(const struct lu_env *env, void *cookie,
369                             lu_printer_t printer,
370                             const struct lu_object_header *hdr)
371 {
372         (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
373                    hdr, hdr->loh_flags, atomic_read(&hdr->loh_ref),
374                    PFID(&hdr->loh_fid),
375                    hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
376                    list_empty((struct list_head *)&hdr->loh_lru) ? "" : " lru",
377                    hdr->loh_attr & LOHA_EXISTS ? " exist":"");
378 }
379 EXPORT_SYMBOL(lu_object_header_print);
380
381 /**
382  * Print human readable representation of the \a o to the \a printer.
383  */
384 void lu_object_print(const struct lu_env *env, void *cookie,
385                      lu_printer_t printer, const struct lu_object *o)
386 {
387         static const char ruler[] = "........................................";
388         struct lu_object_header *top;
389         int depth;
390
391         top = o->lo_header;
392         lu_object_header_print(env, cookie, printer, top);
393         (*printer)(env, cookie, "{ \n");
394         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
395                 depth = o->lo_depth + 4;
396
397                 /*
398                  * print `.' \a depth times followed by type name and address
399                  */
400                 (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
401                            o->lo_dev->ld_type->ldt_name, o);
402                 if (o->lo_ops->loo_object_print != NULL)
403                         o->lo_ops->loo_object_print(env, cookie, printer, o);
404                 (*printer)(env, cookie, "\n");
405         }
406         (*printer)(env, cookie, "} header@%p\n", top);
407 }
408 EXPORT_SYMBOL(lu_object_print);
409
410 /**
411  * Check object consistency.
412  */
413 int lu_object_invariant(const struct lu_object *o)
414 {
415         struct lu_object_header *top;
416
417         top = o->lo_header;
418         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
419                 if (o->lo_ops->loo_object_invariant != NULL &&
420                     !o->lo_ops->loo_object_invariant(o))
421                         return 0;
422         }
423         return 1;
424 }
425 EXPORT_SYMBOL(lu_object_invariant);
426
427 static struct lu_object *htable_lookup(struct lu_site *s,
428                                        const struct hlist_head *bucket,
429                                        const struct lu_fid *f,
430                                        cfs_waitlink_t *waiter)
431 {
432         struct lu_object_header *h;
433         struct hlist_node *scan;
434
435         hlist_for_each_entry(h, scan, bucket, loh_hash) {
436                 s->ls_stats.s_cache_check ++;
437                 if (likely(lu_fid_eq(&h->loh_fid, f))) {
438                         if (unlikely(lu_object_is_dying(h))) {
439                                 /*
440                                  * Lookup found an object being destroyed;
441                                  * this object cannot be returned (to assure
442                                  * that references to dying objects are
443                                  * eventually drained), and moreover, lookup
444                                  * has to wait until object is freed.
445                                  */
446                                 cfs_waitlink_init(waiter);
447                                 cfs_waitq_add(&s->ls_marche_funebre, waiter);
448                                 set_current_state(CFS_TASK_UNINT);
449                                 s->ls_stats.s_cache_death_race ++;
450                                 return ERR_PTR(-EAGAIN);
451                         }
452                         /* bump reference count... */
453                         if (atomic_add_return(1, &h->loh_ref) == 1)
454                                 ++ s->ls_busy;
455                         /* and move to the head of the LRU */
456                         /*
457                          * XXX temporary disable this to measure effects of
458                          * read-write locking.
459                          */
460                         /* list_move_tail(&h->loh_lru, &s->ls_lru); */
461                         s->ls_stats.s_cache_hit ++;
462                         return lu_object_top(h);
463                 }
464         }
465         s->ls_stats.s_cache_miss ++;
466         return NULL;
467 }
468
469 static __u32 fid_hash(const struct lu_fid *f, int bits)
470 {
471         /* all objects with same id and different versions will belong to same
472          * collisions list. */
473         return hash_long(fid_flatten(f), bits);
474 }
475
476 /**
477  * Search cache for an object with the fid \a f. If such object is found,
478  * return it. Otherwise, create new object, insert it into cache and return
479  * it. In any case, additional reference is acquired on the returned object.
480  */
481 struct lu_object *lu_object_find(const struct lu_env *env,
482                                  struct lu_device *dev, const struct lu_fid *f,
483                                  const struct lu_object_conf *conf)
484 {
485         return lu_object_find_at(env, dev->ld_site->ls_top_dev, f, conf);
486 }
487 EXPORT_SYMBOL(lu_object_find);
488
489 /**
490  * Core logic of lu_object_find*() functions.
491  */
492 static struct lu_object *lu_object_find_try(const struct lu_env *env,
493                                             struct lu_device *dev,
494                                             const struct lu_fid *f,
495                                             const struct lu_object_conf *conf,
496                                             cfs_waitlink_t *waiter)
497 {
498         struct lu_site    *s;
499         struct lu_object  *o;
500         struct lu_object  *shadow;
501         struct hlist_head *bucket;
502
503         /*
504          * This uses standard index maintenance protocol:
505          *
506          *     - search index under lock, and return object if found;
507          *     - otherwise, unlock index, allocate new object;
508          *     - lock index and search again;
509          *     - if nothing is found (usual case), insert newly created
510          *       object into index;
511          *     - otherwise (race: other thread inserted object), free
512          *       object just allocated.
513          *     - unlock index;
514          *     - return object.
515          *
516          * If dying object is found during index search, add @waiter to the
517          * site wait-queue and return ERR_PTR(-EAGAIN).
518          */
519
520         s = dev->ld_site;
521         bucket = s->ls_hash + fid_hash(f, s->ls_hash_bits);
522
523         read_lock(&s->ls_guard);
524         o = htable_lookup(s, bucket, f, waiter);
525         read_unlock(&s->ls_guard);
526
527         if (o != NULL)
528                 return o;
529
530         /*
531          * Allocate new object. This may result in rather complicated
532          * operations, including fld queries, inode loading, etc.
533          */
534         o = lu_object_alloc(env, dev, f, conf);
535         if (unlikely(IS_ERR(o)))
536                 return o;
537
538         LASSERT(lu_fid_eq(lu_object_fid(o), f));
539
540         write_lock(&s->ls_guard);
541         shadow = htable_lookup(s, bucket, f, waiter);
542         if (likely(shadow == NULL)) {
543                 hlist_add_head(&o->lo_header->loh_hash, bucket);
544                 list_add_tail(&o->lo_header->loh_lru, &s->ls_lru);
545                 ++ s->ls_busy;
546                 ++ s->ls_total;
547                 shadow = o;
548                 o = NULL;
549         } else
550                 s->ls_stats.s_cache_race ++;
551         write_unlock(&s->ls_guard);
552         if (o != NULL)
553                 lu_object_free(env, o);
554         return shadow;
555 }
556
557 /**
558  * Much like lu_object_find(), but top level device of object is specifically
559  * \a dev rather than top level device of the site. This interface allows
560  * objects of different "stacking" to be created within the same site.
561  */
562 struct lu_object *lu_object_find_at(const struct lu_env *env,
563                                     struct lu_device *dev,
564                                     const struct lu_fid *f,
565                                     const struct lu_object_conf *conf)
566 {
567         struct lu_object *obj;
568         cfs_waitlink_t    wait;
569
570         while (1) {
571                 obj = lu_object_find_try(env, dev, f, conf, &wait);
572                 if (obj == ERR_PTR(-EAGAIN)) {
573                         /*
574                          * lu_object_find_try() already added waiter into the
575                          * wait queue.
576                          */
577                         cfs_waitq_wait(&wait, CFS_TASK_UNINT);
578                         cfs_waitq_del(&dev->ld_site->ls_marche_funebre, &wait);
579                 } else
580                         break;
581         }
582         return obj;
583 }
584 EXPORT_SYMBOL(lu_object_find_at);
585
586 /**
587  * Find object with given fid, and return its slice belonging to given device.
588  */
589 struct lu_object *lu_object_find_slice(const struct lu_env *env,
590                                        struct lu_device *dev,
591                                        const struct lu_fid *f,
592                                        const struct lu_object_conf *conf)
593 {
594         struct lu_object *top;
595         struct lu_object *obj;
596
597         top = lu_object_find(env, dev, f, conf);
598         if (!IS_ERR(top)) {
599                 obj = lu_object_locate(top->lo_header, dev->ld_type);
600                 if (obj == NULL)
601                         lu_object_put(env, top);
602         } else
603                 obj = top;
604         return obj;
605 }
606 EXPORT_SYMBOL(lu_object_find_slice);
607
608 /**
609  * Global list of all device types.
610  */
611 static CFS_LIST_HEAD(lu_device_types);
612
613 int lu_device_type_init(struct lu_device_type *ldt)
614 {
615         int result;
616
617         CFS_INIT_LIST_HEAD(&ldt->ldt_linkage);
618         result = ldt->ldt_ops->ldto_init(ldt);
619         if (result == 0)
620                 list_add(&ldt->ldt_linkage, &lu_device_types);
621         return result;
622 }
623 EXPORT_SYMBOL(lu_device_type_init);
624
625 void lu_device_type_fini(struct lu_device_type *ldt)
626 {
627         list_del_init(&ldt->ldt_linkage);
628         ldt->ldt_ops->ldto_fini(ldt);
629 }
630 EXPORT_SYMBOL(lu_device_type_fini);
631
632 void lu_types_stop(void)
633 {
634         struct lu_device_type *ldt;
635
636         list_for_each_entry(ldt, &lu_device_types, ldt_linkage) {
637                 if (ldt->ldt_device_nr == 0)
638                         ldt->ldt_ops->ldto_stop(ldt);
639         }
640 }
641 EXPORT_SYMBOL(lu_types_stop);
642
643 /**
644  * Global list of all sites on this node
645  */
646 static CFS_LIST_HEAD(lu_sites);
647 static DECLARE_MUTEX(lu_sites_guard);
648
649 /**
650  * Global environment used by site shrinker.
651  */
652 static struct lu_env lu_shrink_env;
653
654 /**
655  * Print all objects in \a s.
656  */
657 void lu_site_print(const struct lu_env *env, struct lu_site *s, void *cookie,
658                    lu_printer_t printer)
659 {
660         int i;
661
662         for (i = 0; i < s->ls_hash_size; ++i) {
663                 struct lu_object_header *h;
664                 struct hlist_node       *scan;
665
666                 read_lock(&s->ls_guard);
667                 hlist_for_each_entry(h, scan, &s->ls_hash[i], loh_hash) {
668
669                         if (!list_empty(&h->loh_layers)) {
670                                 const struct lu_object *obj;
671
672                                 obj = lu_object_top(h);
673                                 lu_object_print(env, cookie, printer, obj);
674                         } else
675                                 lu_object_header_print(env, cookie, printer, h);
676                 }
677                 read_unlock(&s->ls_guard);
678         }
679 }
680 EXPORT_SYMBOL(lu_site_print);
681
682 enum {
683         LU_CACHE_PERCENT   = 20,
684 };
685
686 /**
687  * Return desired hash table order.
688  */
689 static int lu_htable_order(void)
690 {
691         unsigned long cache_size;
692         int bits;
693
694         /*
695          * Calculate hash table size, assuming that we want reasonable
696          * performance when 20% of total memory is occupied by cache of
697          * lu_objects.
698          *
699          * Size of lu_object is (arbitrary) taken as 1K (together with inode).
700          */
701         cache_size = num_physpages;
702
703 #if BITS_PER_LONG == 32
704         /* limit hashtable size for lowmem systems to low RAM */
705         if (cache_size > 1 << (30 - CFS_PAGE_SHIFT))
706                 cache_size = 1 << (30 - CFS_PAGE_SHIFT) * 3 / 4;
707 #endif
708
709         cache_size = cache_size / 100 * LU_CACHE_PERCENT *
710                 (CFS_PAGE_SIZE / 1024);
711
712         for (bits = 1; (1 << bits) < cache_size; ++bits) {
713                 ;
714         }
715         return bits;
716 }
717
718 static struct lock_class_key lu_site_guard_class;
719
720 /**
721  * Initialize site \a s, with \a d as the top level device.
722  */
723 int lu_site_init(struct lu_site *s, struct lu_device *top)
724 {
725         int bits;
726         int size;
727         int i;
728         ENTRY;
729
730         memset(s, 0, sizeof *s);
731         rwlock_init(&s->ls_guard);
732         lockdep_set_class(&s->ls_guard, &lu_site_guard_class);
733         CFS_INIT_LIST_HEAD(&s->ls_lru);
734         CFS_INIT_LIST_HEAD(&s->ls_linkage);
735         cfs_waitq_init(&s->ls_marche_funebre);
736         s->ls_top_dev = top;
737         top->ld_site = s;
738         lu_device_get(top);
739         lu_ref_add(&top->ld_reference, "site-top", s);
740
741         for (bits = lu_htable_order(), size = 1 << bits;
742              (s->ls_hash =
743               cfs_alloc_large(size * sizeof s->ls_hash[0])) == NULL;
744              --bits, size >>= 1) {
745                 /*
746                  * Scale hash table down, until allocation succeeds.
747                  */
748                 ;
749         }
750
751         s->ls_hash_size = size;
752         s->ls_hash_bits = bits;
753         s->ls_hash_mask = size - 1;
754
755         for (i = 0; i < size; i++)
756                 INIT_HLIST_HEAD(&s->ls_hash[i]);
757
758         RETURN(0);
759 }
760 EXPORT_SYMBOL(lu_site_init);
761
762 /**
763  * Finalize \a s and release its resources.
764  */
765 void lu_site_fini(struct lu_site *s)
766 {
767         LASSERT(list_empty(&s->ls_lru));
768         LASSERT(s->ls_total == 0);
769
770         down(&lu_sites_guard);
771         list_del_init(&s->ls_linkage);
772         up(&lu_sites_guard);
773
774         if (s->ls_hash != NULL) {
775                 int i;
776                 for (i = 0; i < s->ls_hash_size; i++)
777                         LASSERT(hlist_empty(&s->ls_hash[i]));
778                 cfs_free_large(s->ls_hash);
779                 s->ls_hash = NULL;
780         }
781         if (s->ls_top_dev != NULL) {
782                 s->ls_top_dev->ld_site = NULL;
783                 lu_ref_del(&s->ls_top_dev->ld_reference, "site-top", s);
784                 lu_device_put(s->ls_top_dev);
785                 s->ls_top_dev = NULL;
786         }
787 }
788 EXPORT_SYMBOL(lu_site_fini);
789
790 /**
791  * Called when initialization of stack for this site is completed.
792  */
793 int lu_site_init_finish(struct lu_site *s)
794 {
795         int result;
796         down(&lu_sites_guard);
797         result = lu_context_refill(&lu_shrink_env.le_ctx);
798         if (result == 0)
799                 list_add(&s->ls_linkage, &lu_sites);
800         up(&lu_sites_guard);
801         return result;
802 }
803 EXPORT_SYMBOL(lu_site_init_finish);
804
805 /**
806  * Acquire additional reference on device \a d
807  */
808 void lu_device_get(struct lu_device *d)
809 {
810         atomic_inc(&d->ld_ref);
811 }
812 EXPORT_SYMBOL(lu_device_get);
813
814 /**
815  * Release reference on device \a d.
816  */
817 void lu_device_put(struct lu_device *d)
818 {
819         LASSERT(atomic_read(&d->ld_ref) > 0);
820         atomic_dec(&d->ld_ref);
821 }
822 EXPORT_SYMBOL(lu_device_put);
823
824 /**
825  * Initialize device \a d of type \a t.
826  */
827 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
828 {
829         if (t->ldt_device_nr++ == 0 && t->ldt_ops->ldto_start != NULL)
830                 t->ldt_ops->ldto_start(t);
831         memset(d, 0, sizeof *d);
832         atomic_set(&d->ld_ref, 0);
833         d->ld_type = t;
834         lu_ref_init(&d->ld_reference);
835         return 0;
836 }
837 EXPORT_SYMBOL(lu_device_init);
838
839 /**
840  * Finalize device \a d.
841  */
842 void lu_device_fini(struct lu_device *d)
843 {
844         struct lu_device_type *t;
845
846         t = d->ld_type;
847         if (d->ld_obd != NULL) {
848                 d->ld_obd->obd_lu_dev = NULL;
849                 d->ld_obd = NULL;
850         }
851
852         lu_ref_fini(&d->ld_reference);
853         LASSERTF(atomic_read(&d->ld_ref) == 0,
854                  "Refcount is %u\n", atomic_read(&d->ld_ref));
855         LASSERT(t->ldt_device_nr > 0);
856         if (--t->ldt_device_nr == 0 && t->ldt_ops->ldto_stop != NULL)
857                 t->ldt_ops->ldto_stop(t);
858 }
859 EXPORT_SYMBOL(lu_device_fini);
860
861 /**
862  * Initialize object \a o that is part of compound object \a h and was created
863  * by device \a d.
864  */
865 int lu_object_init(struct lu_object *o,
866                    struct lu_object_header *h, struct lu_device *d)
867 {
868         memset(o, 0, sizeof *o);
869         o->lo_header = h;
870         o->lo_dev    = d;
871         lu_device_get(d);
872         o->lo_dev_ref = lu_ref_add(&d->ld_reference, "lu_object", o);
873         CFS_INIT_LIST_HEAD(&o->lo_linkage);
874         return 0;
875 }
876 EXPORT_SYMBOL(lu_object_init);
877
878 /**
879  * Finalize object and release its resources.
880  */
881 void lu_object_fini(struct lu_object *o)
882 {
883         struct lu_device *dev = o->lo_dev;
884
885         LASSERT(list_empty(&o->lo_linkage));
886
887         if (dev != NULL) {
888                 lu_ref_del_at(&dev->ld_reference,
889                               o->lo_dev_ref , "lu_object", o);
890                 lu_device_put(dev);
891                 o->lo_dev = NULL;
892         }
893 }
894 EXPORT_SYMBOL(lu_object_fini);
895
896 /**
897  * Add object \a o as first layer of compound object \a h
898  *
899  * This is typically called by the ->ldo_object_alloc() method of top-level
900  * device.
901  */
902 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
903 {
904         list_move(&o->lo_linkage, &h->loh_layers);
905 }
906 EXPORT_SYMBOL(lu_object_add_top);
907
908 /**
909  * Add object \a o as a layer of compound object, going after \a before.
910  *
911  * This is typically called by the ->ldo_object_alloc() method of \a
912  * before->lo_dev.
913  */
914 void lu_object_add(struct lu_object *before, struct lu_object *o)
915 {
916         list_move(&o->lo_linkage, &before->lo_linkage);
917 }
918 EXPORT_SYMBOL(lu_object_add);
919
920 /**
921  * Initialize compound object.
922  */
923 int lu_object_header_init(struct lu_object_header *h)
924 {
925         memset(h, 0, sizeof *h);
926         atomic_set(&h->loh_ref, 1);
927         INIT_HLIST_NODE(&h->loh_hash);
928         CFS_INIT_LIST_HEAD(&h->loh_lru);
929         CFS_INIT_LIST_HEAD(&h->loh_layers);
930         lu_ref_init(&h->loh_reference);
931         return 0;
932 }
933 EXPORT_SYMBOL(lu_object_header_init);
934
935 /**
936  * Finalize compound object.
937  */
938 void lu_object_header_fini(struct lu_object_header *h)
939 {
940         LASSERT(list_empty(&h->loh_layers));
941         LASSERT(list_empty(&h->loh_lru));
942         LASSERT(hlist_unhashed(&h->loh_hash));
943         lu_ref_fini(&h->loh_reference);
944 }
945 EXPORT_SYMBOL(lu_object_header_fini);
946
947 /**
948  * Given a compound object, find its slice, corresponding to the device type
949  * \a dtype.
950  */
951 struct lu_object *lu_object_locate(struct lu_object_header *h,
952                                    const struct lu_device_type *dtype)
953 {
954         struct lu_object *o;
955
956         list_for_each_entry(o, &h->loh_layers, lo_linkage) {
957                 if (o->lo_dev->ld_type == dtype)
958                         return o;
959         }
960         return NULL;
961 }
962 EXPORT_SYMBOL(lu_object_locate);
963
964
965
966 /**
967  * Finalize and free devices in the device stack.
968  *
969  * Finalize device stack by purging object cache, and calling
970  * lu_device_type_operations::ldto_device_fini() and
971  * lu_device_type_operations::ldto_device_free() on all devices in the stack.
972  */
973 void lu_stack_fini(const struct lu_env *env, struct lu_device *top)
974 {
975         struct lu_site   *site = top->ld_site;
976         struct lu_device *scan;
977         struct lu_device *next;
978
979         lu_site_purge(env, site, ~0);
980         for (scan = top; scan != NULL; scan = next) {
981                 next = scan->ld_type->ldt_ops->ldto_device_fini(env, scan);
982                 lu_ref_del(&scan->ld_reference, "lu-stack", &lu_site_init);
983                 lu_device_put(scan);
984         }
985
986         /* purge again. */
987         lu_site_purge(env, site, ~0);
988
989         if (!list_empty(&site->ls_lru) || site->ls_total != 0) {
990                 /*
991                  * Uh-oh, objects still exist.
992                  */
993                 static DECLARE_LU_CDEBUG_PRINT_INFO(cookie, D_ERROR);
994
995                 lu_site_print(env, site, &cookie, lu_cdebug_printer);
996         }
997
998         for (scan = top; scan != NULL; scan = next) {
999                 const struct lu_device_type *ldt = scan->ld_type;
1000                 struct obd_type             *type;
1001
1002                 next = ldt->ldt_ops->ldto_device_free(env, scan);
1003                 type = ldt->ldt_obd_type;
1004                 if (type != NULL) {
1005                         type->typ_refcnt--;
1006                         class_put_type(type);
1007                 }
1008         }
1009 }
1010 EXPORT_SYMBOL(lu_stack_fini);
1011
1012 enum {
1013         /**
1014          * Maximal number of tld slots.
1015          */
1016         LU_CONTEXT_KEY_NR = 32
1017 };
1018
1019 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
1020
1021 static spinlock_t lu_keys_guard = SPIN_LOCK_UNLOCKED;
1022
1023 /**
1024  * Global counter incremented whenever key is registered, unregistered,
1025  * revived or quiesced. This is used to void unnecessary calls to
1026  * lu_context_refill(). No locking is provided, as initialization and shutdown
1027  * are supposed to be externally serialized.
1028  */
1029 static unsigned key_set_version = 0;
1030
1031 /**
1032  * Register new key.
1033  */
1034 int lu_context_key_register(struct lu_context_key *key)
1035 {
1036         int result;
1037         int i;
1038
1039         LASSERT(key->lct_init != NULL);
1040         LASSERT(key->lct_fini != NULL);
1041         LASSERT(key->lct_tags != 0);
1042         LASSERT(key->lct_owner != NULL);
1043
1044         result = -ENFILE;
1045         spin_lock(&lu_keys_guard);
1046         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1047                 if (lu_keys[i] == NULL) {
1048                         key->lct_index = i;
1049                         atomic_set(&key->lct_used, 1);
1050                         lu_keys[i] = key;
1051                         lu_ref_init(&key->lct_reference);
1052                         result = 0;
1053                         ++key_set_version;
1054                         break;
1055                 }
1056         }
1057         spin_unlock(&lu_keys_guard);
1058         return result;
1059 }
1060 EXPORT_SYMBOL(lu_context_key_register);
1061
1062 static void key_fini(struct lu_context *ctx, int index)
1063 {
1064         if (ctx->lc_value != NULL && ctx->lc_value[index] != NULL) {
1065                 struct lu_context_key *key;
1066
1067                 key = lu_keys[index];
1068                 LASSERT(key != NULL);
1069                 LASSERT(key->lct_fini != NULL);
1070                 LASSERT(atomic_read(&key->lct_used) > 1);
1071
1072                 key->lct_fini(ctx, key, ctx->lc_value[index]);
1073                 lu_ref_del(&key->lct_reference, "ctx", ctx);
1074                 atomic_dec(&key->lct_used);
1075                 LASSERT(key->lct_owner != NULL);
1076                 if (!(ctx->lc_tags & LCT_NOREF)) {
1077                         LASSERT(module_refcount(key->lct_owner) > 0);
1078                         module_put(key->lct_owner);
1079                 }
1080                 ctx->lc_value[index] = NULL;
1081         }
1082 }
1083
1084 /**
1085  * Deregister key.
1086  */
1087 void lu_context_key_degister(struct lu_context_key *key)
1088 {
1089         LASSERT(atomic_read(&key->lct_used) >= 1);
1090         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1091
1092         lu_context_key_quiesce(key);
1093
1094         ++key_set_version;
1095         spin_lock(&lu_keys_guard);
1096         key_fini(&lu_shrink_env.le_ctx, key->lct_index);
1097         lu_keys[key->lct_index] = NULL;
1098         spin_unlock(&lu_keys_guard);
1099
1100         LASSERTF(atomic_read(&key->lct_used) == 1, "key has instances: %d\n",
1101                  atomic_read(&key->lct_used));
1102 }
1103 EXPORT_SYMBOL(lu_context_key_degister);
1104
1105 /**
1106  * Register a number of keys. This has to be called after all keys have been
1107  * initialized by a call to LU_CONTEXT_KEY_INIT().
1108  */
1109 int lu_context_key_register_many(struct lu_context_key *k, ...)
1110 {
1111         struct lu_context_key *key = k;
1112         va_list args;
1113         int result;
1114
1115         va_start(args, k);
1116         do {
1117                 result = lu_context_key_register(key);
1118                 if (result)
1119                         break;
1120                 key = va_arg(args, struct lu_context_key *);
1121         } while (key != NULL);
1122         va_end(args);
1123
1124         if (result != 0) {
1125                 va_start(args, k);
1126                 while (k != key) {
1127                         lu_context_key_degister(k);
1128                         k = va_arg(args, struct lu_context_key *);
1129                 }
1130                 va_end(args);
1131         }
1132
1133         return result;
1134 }
1135 EXPORT_SYMBOL(lu_context_key_register_many);
1136
1137 /**
1138  * De-register a number of keys. This is a dual to
1139  * lu_context_key_register_many().
1140  */
1141 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1142 {
1143         va_list args;
1144
1145         va_start(args, k);
1146         do {
1147                 lu_context_key_degister(k);
1148                 k = va_arg(args, struct lu_context_key*);
1149         } while (k != NULL);
1150         va_end(args);
1151 }
1152 EXPORT_SYMBOL(lu_context_key_degister_many);
1153
1154 /**
1155  * Revive a number of keys.
1156  */
1157 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1158 {
1159         va_list args;
1160
1161         va_start(args, k);
1162         do {
1163                 lu_context_key_revive(k);
1164                 k = va_arg(args, struct lu_context_key*);
1165         } while (k != NULL);
1166         va_end(args);
1167 }
1168 EXPORT_SYMBOL(lu_context_key_revive_many);
1169
1170 /**
1171  * Quiescent a number of keys.
1172  */
1173 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1174 {
1175         va_list args;
1176
1177         va_start(args, k);
1178         do {
1179                 lu_context_key_quiesce(k);
1180                 k = va_arg(args, struct lu_context_key*);
1181         } while (k != NULL);
1182         va_end(args);
1183 }
1184 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1185
1186 /**
1187  * Return value associated with key \a key in context \a ctx.
1188  */
1189 void *lu_context_key_get(const struct lu_context *ctx,
1190                          const struct lu_context_key *key)
1191 {
1192         LINVRNT(ctx->lc_state == LCS_ENTERED);
1193         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1194         LASSERT(lu_keys[key->lct_index] == key);
1195         return ctx->lc_value[key->lct_index];
1196 }
1197 EXPORT_SYMBOL(lu_context_key_get);
1198
1199 /**
1200  * List of remembered contexts. XXX document me.
1201  */
1202 static CFS_LIST_HEAD(lu_context_remembered);
1203
1204 /**
1205  * Destroy \a key in all remembered contexts. This is used to destroy key
1206  * values in "shared" contexts (like service threads), when a module owning
1207  * the key is about to be unloaded.
1208  */
1209 void lu_context_key_quiesce(struct lu_context_key *key)
1210 {
1211         struct lu_context *ctx;
1212         extern unsigned cl_env_cache_purge(unsigned nr);
1213
1214         if (!(key->lct_tags & LCT_QUIESCENT)) {
1215                 /*
1216                  * XXX layering violation.
1217                  */
1218                 cl_env_cache_purge(~0);
1219                 key->lct_tags |= LCT_QUIESCENT;
1220                 /*
1221                  * XXX memory barrier has to go here.
1222                  */
1223                 spin_lock(&lu_keys_guard);
1224                 list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
1225                         key_fini(ctx, key->lct_index);
1226                 spin_unlock(&lu_keys_guard);
1227                 ++key_set_version;
1228         }
1229 }
1230 EXPORT_SYMBOL(lu_context_key_quiesce);
1231
1232 void lu_context_key_revive(struct lu_context_key *key)
1233 {
1234         key->lct_tags &= ~LCT_QUIESCENT;
1235         ++key_set_version;
1236 }
1237 EXPORT_SYMBOL(lu_context_key_revive);
1238
1239 static void keys_fini(struct lu_context *ctx)
1240 {
1241         int i;
1242
1243         spin_lock(&lu_keys_guard);
1244         if (ctx->lc_value != NULL) {
1245                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1246                         key_fini(ctx, i);
1247                 OBD_FREE(ctx->lc_value,
1248                          ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1249                 ctx->lc_value = NULL;
1250         }
1251         spin_unlock(&lu_keys_guard);
1252 }
1253
1254 static int keys_fill(struct lu_context *ctx)
1255 {
1256         int i;
1257
1258         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1259                 struct lu_context_key *key;
1260
1261                 key = lu_keys[i];
1262                 if (ctx->lc_value[i] == NULL && key != NULL &&
1263                     (key->lct_tags & ctx->lc_tags) &&
1264                     /*
1265                      * Don't create values for a LCT_QUIESCENT key, as this
1266                      * will pin module owning a key.
1267                      */
1268                     !(key->lct_tags & LCT_QUIESCENT)) {
1269                         void *value;
1270
1271                         LINVRNT(key->lct_init != NULL);
1272                         LINVRNT(key->lct_index == i);
1273
1274                         value = key->lct_init(ctx, key);
1275                         if (unlikely(IS_ERR(value)))
1276                                 return PTR_ERR(value);
1277
1278                         LASSERT(key->lct_owner != NULL);
1279                         if (!(ctx->lc_tags & LCT_NOREF))
1280                                 try_module_get(key->lct_owner);
1281                         lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1282                         atomic_inc(&key->lct_used);
1283                         /*
1284                          * This is the only place in the code, where an
1285                          * element of ctx->lc_value[] array is set to non-NULL
1286                          * value.
1287                          */
1288                         ctx->lc_value[i] = value;
1289                         if (key->lct_exit != NULL)
1290                                 ctx->lc_tags |= LCT_HAS_EXIT;
1291                 }
1292                 ctx->lc_version = key_set_version;
1293         }
1294         return 0;
1295 }
1296
1297 static int keys_init(struct lu_context *ctx)
1298 {
1299         int result;
1300
1301         OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1302         if (likely(ctx->lc_value != NULL))
1303                 result = keys_fill(ctx);
1304         else
1305                 result = -ENOMEM;
1306
1307         if (result != 0)
1308                 keys_fini(ctx);
1309         return result;
1310 }
1311
1312 /**
1313  * Initialize context data-structure. Create values for all keys.
1314  */
1315 int lu_context_init(struct lu_context *ctx, __u32 tags)
1316 {
1317         memset(ctx, 0, sizeof *ctx);
1318         ctx->lc_state = LCS_INITIALIZED;
1319         ctx->lc_tags = tags;
1320         if (tags & LCT_REMEMBER) {
1321                 spin_lock(&lu_keys_guard);
1322                 list_add(&ctx->lc_remember, &lu_context_remembered);
1323                 spin_unlock(&lu_keys_guard);
1324         } else
1325                 CFS_INIT_LIST_HEAD(&ctx->lc_remember);
1326         return keys_init(ctx);
1327 }
1328 EXPORT_SYMBOL(lu_context_init);
1329
1330 /**
1331  * Finalize context data-structure. Destroy key values.
1332  */
1333 void lu_context_fini(struct lu_context *ctx)
1334 {
1335         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1336         ctx->lc_state = LCS_FINALIZED;
1337         keys_fini(ctx);
1338         spin_lock(&lu_keys_guard);
1339         list_del_init(&ctx->lc_remember);
1340         spin_unlock(&lu_keys_guard);
1341 }
1342 EXPORT_SYMBOL(lu_context_fini);
1343
1344 /**
1345  * Called before entering context.
1346  */
1347 void lu_context_enter(struct lu_context *ctx)
1348 {
1349         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1350         ctx->lc_state = LCS_ENTERED;
1351 }
1352 EXPORT_SYMBOL(lu_context_enter);
1353
1354 /**
1355  * Called after exiting from \a ctx
1356  */
1357 void lu_context_exit(struct lu_context *ctx)
1358 {
1359         int i;
1360
1361         LINVRNT(ctx->lc_state == LCS_ENTERED);
1362         ctx->lc_state = LCS_LEFT;
1363         if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
1364                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1365                         if (ctx->lc_value[i] != NULL) {
1366                                 struct lu_context_key *key;
1367
1368                                 key = lu_keys[i];
1369                                 LASSERT(key != NULL);
1370                                 if (key->lct_exit != NULL)
1371                                         key->lct_exit(ctx,
1372                                                       key, ctx->lc_value[i]);
1373                         }
1374                 }
1375         }
1376 }
1377 EXPORT_SYMBOL(lu_context_exit);
1378
1379 /**
1380  * Allocate for context all missing keys that were registered after context
1381  * creation.
1382  */
1383 int lu_context_refill(struct lu_context *ctx)
1384 {
1385         LINVRNT(ctx->lc_value != NULL);
1386         return ctx->lc_version == key_set_version ? 0 : keys_fill(ctx);
1387 }
1388 EXPORT_SYMBOL(lu_context_refill);
1389
1390 int lu_env_init(struct lu_env *env, __u32 tags)
1391 {
1392         int result;
1393
1394         env->le_ses = NULL;
1395         result = lu_context_init(&env->le_ctx, tags);
1396         if (likely(result == 0))
1397                 lu_context_enter(&env->le_ctx);
1398         return result;
1399 }
1400 EXPORT_SYMBOL(lu_env_init);
1401
1402 void lu_env_fini(struct lu_env *env)
1403 {
1404         lu_context_exit(&env->le_ctx);
1405         lu_context_fini(&env->le_ctx);
1406         env->le_ses = NULL;
1407 }
1408 EXPORT_SYMBOL(lu_env_fini);
1409
1410 int lu_env_refill(struct lu_env *env)
1411 {
1412         int result;
1413
1414         result = lu_context_refill(&env->le_ctx);
1415         if (result == 0 && env->le_ses != NULL)
1416                 result = lu_context_refill(env->le_ses);
1417         return result;
1418 }
1419 EXPORT_SYMBOL(lu_env_refill);
1420
1421 static struct shrinker *lu_site_shrinker = NULL;
1422
1423 #ifdef __KERNEL__
1424 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1425 {
1426         struct lu_site *s;
1427         struct lu_site *tmp;
1428         int cached = 0;
1429         int remain = nr;
1430         CFS_LIST_HEAD(splice);
1431
1432         if (nr != 0) {
1433                 if (!(gfp_mask & __GFP_FS))
1434                         return -1;
1435                 CDEBUG(D_INODE, "Shrink %d objects\n", nr);
1436         }
1437
1438         down(&lu_sites_guard);
1439         list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1440                 if (nr != 0) {
1441                         remain = lu_site_purge(&lu_shrink_env, s, remain);
1442                         /*
1443                          * Move just shrunk site to the tail of site list to
1444                          * assure shrinking fairness.
1445                          */
1446                         list_move_tail(&s->ls_linkage, &splice);
1447                 }
1448                 read_lock(&s->ls_guard);
1449                 cached += s->ls_total - s->ls_busy;
1450                 read_unlock(&s->ls_guard);
1451                 if (nr && remain <= 0)
1452                         break;
1453         }
1454         list_splice(&splice, lu_sites.prev);
1455         up(&lu_sites_guard);
1456
1457         cached = (cached / 100) * sysctl_vfs_cache_pressure;
1458         if (nr == 0)
1459                 CDEBUG(D_INODE, "%d objects cached\n", cached);
1460         return cached;
1461 }
1462
1463 /*
1464  * Debugging stuff.
1465  */
1466
1467 /**
1468  * Environment to be used in debugger, contains all tags.
1469  */
1470 struct lu_env lu_debugging_env;
1471
1472 /**
1473  * Debugging printer function using printk().
1474  */
1475 int lu_printk_printer(const struct lu_env *env,
1476                       void *_, const char *format, ...)
1477 {
1478         va_list args;
1479
1480         va_start(args, format);
1481         vprintk(format, args);
1482         va_end(args);
1483         return 0;
1484 }
1485
1486 void lu_debugging_setup(void)
1487 {
1488         lu_env_init(&lu_debugging_env, ~0);
1489 }
1490
1491 void lu_context_keys_dump(void)
1492 {
1493         int i;
1494
1495         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1496                 struct lu_context_key *key;
1497
1498                 key = lu_keys[i];
1499                 if (key != NULL) {
1500                         CERROR("[%i]: %p %x (%p,%p,%p) %i %i \"%s\"@%p\n",
1501                                i, key, key->lct_tags,
1502                                key->lct_init, key->lct_fini, key->lct_exit,
1503                                key->lct_index, atomic_read(&key->lct_used),
1504                                key->lct_owner ? key->lct_owner->name : "",
1505                                key->lct_owner);
1506                         lu_ref_print(&key->lct_reference);
1507                 }
1508         }
1509 }
1510 EXPORT_SYMBOL(lu_context_keys_dump);
1511 #else  /* !__KERNEL__ */
1512 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1513 {
1514         return 0;
1515 }
1516 #endif /* __KERNEL__ */
1517
1518 int  cl_global_init(void);
1519 void cl_global_fini(void);
1520 int  lu_ref_global_init(void);
1521 void lu_ref_global_fini(void);
1522
1523 int dt_global_init(void);
1524 void dt_global_fini(void);
1525
1526 int llo_global_init(void);
1527 void llo_global_fini(void);
1528
1529 /**
1530  * Initialization of global lu_* data.
1531  */
1532 int lu_global_init(void)
1533 {
1534         int result;
1535
1536         CDEBUG(D_CONSOLE, "Lustre LU module (%p).\n", &lu_keys);
1537
1538         LU_CONTEXT_KEY_INIT(&lu_global_key);
1539         result = lu_context_key_register(&lu_global_key);
1540         if (result != 0)
1541                 return result;
1542         /*
1543          * At this level, we don't know what tags are needed, so allocate them
1544          * conservatively. This should not be too bad, because this
1545          * environment is global.
1546          */
1547         down(&lu_sites_guard);
1548         result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
1549         up(&lu_sites_guard);
1550         if (result != 0)
1551                 return result;
1552
1553         result = lu_ref_global_init();
1554         if (result != 0)
1555                 return result;
1556         /*
1557          * seeks estimation: 3 seeks to read a record from oi, one to read
1558          * inode, one for ea. Unfortunately setting this high value results in
1559          * lu_object/inode cache consuming all the memory.
1560          */
1561         lu_site_shrinker = set_shrinker(DEFAULT_SEEKS, lu_cache_shrink);
1562         if (lu_site_shrinker == NULL)
1563                 return -ENOMEM;
1564
1565         result = lu_time_global_init();
1566         if (result)
1567                 GOTO(out, result);
1568
1569 #ifdef __KERNEL__
1570         result = dt_global_init();
1571         if (result)
1572                 GOTO(out, result);
1573
1574         result = llo_global_init();
1575         if (result)
1576                 GOTO(out, result);
1577 #endif
1578         result = cl_global_init();
1579 out:
1580
1581         return result;
1582 }
1583
1584 /**
1585  * Dual to lu_global_init().
1586  */
1587 void lu_global_fini(void)
1588 {
1589         cl_global_fini();
1590 #ifdef __KERNEL__
1591         llo_global_fini();
1592         dt_global_fini();
1593 #endif
1594         lu_time_global_fini();
1595         if (lu_site_shrinker != NULL) {
1596                 remove_shrinker(lu_site_shrinker);
1597                 lu_site_shrinker = NULL;
1598         }
1599
1600         lu_context_key_degister(&lu_global_key);
1601
1602         /*
1603          * Tear shrinker environment down _after_ de-registering
1604          * lu_global_key, because the latter has a value in the former.
1605          */
1606         down(&lu_sites_guard);
1607         lu_env_fini(&lu_shrink_env);
1608         up(&lu_sites_guard);
1609
1610         lu_ref_global_fini();
1611 }
1612
1613 struct lu_buf LU_BUF_NULL = {
1614         .lb_buf = NULL,
1615         .lb_len = 0
1616 };
1617 EXPORT_SYMBOL(LU_BUF_NULL);
1618
1619 /**
1620  * Output site statistical counters into a buffer. Suitable for
1621  * lprocfs_rd_*()-style functions.
1622  */
1623 int lu_site_stats_print(const struct lu_site *s, char *page, int count)
1624 {
1625         int i;
1626         int populated;
1627
1628         /*
1629          * How many hash buckets are not-empty? Don't bother with locks: it's
1630          * an estimation anyway.
1631          */
1632         for (i = 0, populated = 0; i < s->ls_hash_size; i++)
1633                 populated += !hlist_empty(&s->ls_hash[i]);
1634
1635         return snprintf(page, count, "%d %d %d/%d %d %d %d %d %d %d %d\n",
1636                         s->ls_total,
1637                         s->ls_busy,
1638                         populated,
1639                         s->ls_hash_size,
1640                         s->ls_stats.s_created,
1641                         s->ls_stats.s_cache_hit,
1642                         s->ls_stats.s_cache_miss,
1643                         s->ls_stats.s_cache_check,
1644                         s->ls_stats.s_cache_race,
1645                         s->ls_stats.s_cache_death_race,
1646                         s->ls_stats.s_lru_purged);
1647 }
1648 EXPORT_SYMBOL(lu_site_stats_print);
1649
1650 #ifdef __KERNEL__
1651 /*
1652  * XXX: Functions below logically belong to the fid module, but they are used
1653  * by dt_store_open(). Put them here until better place is found.
1654  */
1655
1656 void fid_pack(struct lu_fid_pack *pack, const struct lu_fid *fid,
1657               struct lu_fid *befider)
1658 {
1659         int recsize;
1660         __u64 seq;
1661         __u32 oid;
1662
1663         seq = fid_seq(fid);
1664         oid = fid_oid(fid);
1665
1666         /*
1667          * Two cases: compact 6 bytes representation for a common case, and
1668          * full 17 byte representation for "unusual" fid.
1669          */
1670
1671         /*
1672          * Check that usual case is really usual.
1673          */
1674         CLASSERT(LUSTRE_SEQ_MAX_WIDTH < 0xffffull);
1675
1676         /* fid can be packed in six bytes (first byte as length of packed fid,
1677          * three bytes of seq and two bytes of oid).
1678          * this reduces IO overhead specially for OSD Object Index. */
1679
1680         if (seq < FID_SEQ_START ||
1681             seq > (0xffffffull + FID_SEQ_START) ||
1682             oid > 0xffff || fid_ver(fid) != 0) {
1683                 fid_cpu_to_be(befider, fid);
1684                 recsize = sizeof *befider;
1685         } else {
1686                 unsigned char *small_befider;
1687
1688                 small_befider = (unsigned char *)befider;
1689
1690                 small_befider[0] = seq >> 16;
1691                 small_befider[1] = seq >> 8;
1692                 small_befider[2] = seq;
1693
1694                 small_befider[3] = oid >> 8;
1695                 small_befider[4] = oid;
1696
1697                 recsize = 5;
1698         }
1699         memcpy(pack->fp_area, befider, recsize);
1700         pack->fp_len = recsize + 1;
1701 }
1702 EXPORT_SYMBOL(fid_pack);
1703
1704 int fid_unpack(const struct lu_fid_pack *pack, struct lu_fid *fid)
1705 {
1706         int result;
1707
1708         result = 0;
1709         switch (pack->fp_len) {
1710         case sizeof *fid + 1:
1711                 memcpy(fid, pack->fp_area, sizeof *fid);
1712                 fid_be_to_cpu(fid, fid);
1713                 break;
1714         case 6: {
1715                 const unsigned char *area;
1716
1717                 area = (unsigned char *)pack->fp_area;
1718                 fid->f_seq = (area[0] << 16) | (area[1] << 8) | area[2];
1719                 fid->f_seq +=  FID_SEQ_START;
1720                 fid->f_oid = (area[3] << 8) | area[4];
1721                 fid->f_ver = 0;
1722                 break;
1723         }
1724         default:
1725                 CERROR("Unexpected packed fid size: %d\n", pack->fp_len);
1726                 result = -EIO;
1727         }
1728         return result;
1729 }
1730 EXPORT_SYMBOL(fid_unpack);
1731 #endif  /* #ifdef __KERNEL__ */
1732
1733 const char *lu_time_names[LU_TIME_NR] = {
1734         [LU_TIME_FIND_LOOKUP] = "find_lookup",
1735         [LU_TIME_FIND_ALLOC]  = "find_alloc",
1736         [LU_TIME_FIND_INSERT] = "find_insert"
1737 };
1738 EXPORT_SYMBOL(lu_time_names);
1739
1740 /**
1741  * Helper function to initialize a number of kmem slab caches at once.
1742  */
1743 int lu_kmem_init(struct lu_kmem_descr *caches)
1744 {
1745         int result;
1746
1747         for (result = 0; caches->ckd_cache != NULL; ++caches) {
1748                 *caches->ckd_cache = cfs_mem_cache_create(caches->ckd_name,
1749                                                           caches->ckd_size,
1750                                                           0, 0);
1751                 if (*caches->ckd_cache == NULL) {
1752                         result = -ENOMEM;
1753                         break;
1754                 }
1755         }
1756         return result;
1757 }
1758 EXPORT_SYMBOL(lu_kmem_init);
1759
1760 /**
1761  * Helper function to finalize a number of kmem slab cached at once. Dual to
1762  * lu_kmem_init().
1763  */
1764 void lu_kmem_fini(struct lu_kmem_descr *caches)
1765 {
1766         int rc;
1767
1768         for (; caches->ckd_cache != NULL; ++caches) {
1769                 if (*caches->ckd_cache != NULL) {
1770                         rc = cfs_mem_cache_destroy(*caches->ckd_cache);
1771                         LASSERTF(rc == 0, "couldn't destroy %s slab\n",
1772                                  caches->ckd_name);
1773                         *caches->ckd_cache = NULL;
1774                 }
1775         }
1776 }
1777 EXPORT_SYMBOL(lu_kmem_fini);