Whamcloud - gitweb
*** empty log message ***
[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         if (lu_keys[key->lct_index]) {
1098                 lu_keys[key->lct_index] = NULL;
1099                 lu_ref_fini(&key->lct_reference);
1100         }
1101         spin_unlock(&lu_keys_guard);
1102
1103         LASSERTF(atomic_read(&key->lct_used) == 1, "key has instances: %d\n",
1104                  atomic_read(&key->lct_used));
1105 }
1106 EXPORT_SYMBOL(lu_context_key_degister);
1107
1108 /**
1109  * Register a number of keys. This has to be called after all keys have been
1110  * initialized by a call to LU_CONTEXT_KEY_INIT().
1111  */
1112 int lu_context_key_register_many(struct lu_context_key *k, ...)
1113 {
1114         struct lu_context_key *key = k;
1115         va_list args;
1116         int result;
1117
1118         va_start(args, k);
1119         do {
1120                 result = lu_context_key_register(key);
1121                 if (result)
1122                         break;
1123                 key = va_arg(args, struct lu_context_key *);
1124         } while (key != NULL);
1125         va_end(args);
1126
1127         if (result != 0) {
1128                 va_start(args, k);
1129                 while (k != key) {
1130                         lu_context_key_degister(k);
1131                         k = va_arg(args, struct lu_context_key *);
1132                 }
1133                 va_end(args);
1134         }
1135
1136         return result;
1137 }
1138 EXPORT_SYMBOL(lu_context_key_register_many);
1139
1140 /**
1141  * De-register a number of keys. This is a dual to
1142  * lu_context_key_register_many().
1143  */
1144 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1145 {
1146         va_list args;
1147
1148         va_start(args, k);
1149         do {
1150                 lu_context_key_degister(k);
1151                 k = va_arg(args, struct lu_context_key*);
1152         } while (k != NULL);
1153         va_end(args);
1154 }
1155 EXPORT_SYMBOL(lu_context_key_degister_many);
1156
1157 /**
1158  * Revive a number of keys.
1159  */
1160 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1161 {
1162         va_list args;
1163
1164         va_start(args, k);
1165         do {
1166                 lu_context_key_revive(k);
1167                 k = va_arg(args, struct lu_context_key*);
1168         } while (k != NULL);
1169         va_end(args);
1170 }
1171 EXPORT_SYMBOL(lu_context_key_revive_many);
1172
1173 /**
1174  * Quiescent a number of keys.
1175  */
1176 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1177 {
1178         va_list args;
1179
1180         va_start(args, k);
1181         do {
1182                 lu_context_key_quiesce(k);
1183                 k = va_arg(args, struct lu_context_key*);
1184         } while (k != NULL);
1185         va_end(args);
1186 }
1187 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1188
1189 /**
1190  * Return value associated with key \a key in context \a ctx.
1191  */
1192 void *lu_context_key_get(const struct lu_context *ctx,
1193                          const struct lu_context_key *key)
1194 {
1195         LINVRNT(ctx->lc_state == LCS_ENTERED);
1196         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1197         LASSERT(lu_keys[key->lct_index] == key);
1198         return ctx->lc_value[key->lct_index];
1199 }
1200 EXPORT_SYMBOL(lu_context_key_get);
1201
1202 /**
1203  * List of remembered contexts. XXX document me.
1204  */
1205 static CFS_LIST_HEAD(lu_context_remembered);
1206
1207 /**
1208  * Destroy \a key in all remembered contexts. This is used to destroy key
1209  * values in "shared" contexts (like service threads), when a module owning
1210  * the key is about to be unloaded.
1211  */
1212 void lu_context_key_quiesce(struct lu_context_key *key)
1213 {
1214         struct lu_context *ctx;
1215         extern unsigned cl_env_cache_purge(unsigned nr);
1216
1217         if (!(key->lct_tags & LCT_QUIESCENT)) {
1218                 /*
1219                  * XXX layering violation.
1220                  */
1221                 cl_env_cache_purge(~0);
1222                 key->lct_tags |= LCT_QUIESCENT;
1223                 /*
1224                  * XXX memory barrier has to go here.
1225                  */
1226                 spin_lock(&lu_keys_guard);
1227                 list_for_each_entry(ctx, &lu_context_remembered, lc_remember)
1228                         key_fini(ctx, key->lct_index);
1229                 spin_unlock(&lu_keys_guard);
1230                 ++key_set_version;
1231         }
1232 }
1233 EXPORT_SYMBOL(lu_context_key_quiesce);
1234
1235 void lu_context_key_revive(struct lu_context_key *key)
1236 {
1237         key->lct_tags &= ~LCT_QUIESCENT;
1238         ++key_set_version;
1239 }
1240 EXPORT_SYMBOL(lu_context_key_revive);
1241
1242 static void keys_fini(struct lu_context *ctx)
1243 {
1244         int i;
1245
1246         spin_lock(&lu_keys_guard);
1247         if (ctx->lc_value != NULL) {
1248                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1249                         key_fini(ctx, i);
1250                 OBD_FREE(ctx->lc_value,
1251                          ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1252                 ctx->lc_value = NULL;
1253         }
1254         spin_unlock(&lu_keys_guard);
1255 }
1256
1257 static int keys_fill(struct lu_context *ctx)
1258 {
1259         int i;
1260
1261         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1262                 struct lu_context_key *key;
1263
1264                 key = lu_keys[i];
1265                 if (ctx->lc_value[i] == NULL && key != NULL &&
1266                     (key->lct_tags & ctx->lc_tags) &&
1267                     /*
1268                      * Don't create values for a LCT_QUIESCENT key, as this
1269                      * will pin module owning a key.
1270                      */
1271                     !(key->lct_tags & LCT_QUIESCENT)) {
1272                         void *value;
1273
1274                         LINVRNT(key->lct_init != NULL);
1275                         LINVRNT(key->lct_index == i);
1276
1277                         value = key->lct_init(ctx, key);
1278                         if (unlikely(IS_ERR(value)))
1279                                 return PTR_ERR(value);
1280
1281                         LASSERT(key->lct_owner != NULL);
1282                         if (!(ctx->lc_tags & LCT_NOREF))
1283                                 try_module_get(key->lct_owner);
1284                         lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1285                         atomic_inc(&key->lct_used);
1286                         /*
1287                          * This is the only place in the code, where an
1288                          * element of ctx->lc_value[] array is set to non-NULL
1289                          * value.
1290                          */
1291                         ctx->lc_value[i] = value;
1292                         if (key->lct_exit != NULL)
1293                                 ctx->lc_tags |= LCT_HAS_EXIT;
1294                 }
1295                 ctx->lc_version = key_set_version;
1296         }
1297         return 0;
1298 }
1299
1300 static int keys_init(struct lu_context *ctx)
1301 {
1302         int result;
1303
1304         OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1305         if (likely(ctx->lc_value != NULL))
1306                 result = keys_fill(ctx);
1307         else
1308                 result = -ENOMEM;
1309
1310         if (result != 0)
1311                 keys_fini(ctx);
1312         return result;
1313 }
1314
1315 /**
1316  * Initialize context data-structure. Create values for all keys.
1317  */
1318 int lu_context_init(struct lu_context *ctx, __u32 tags)
1319 {
1320         memset(ctx, 0, sizeof *ctx);
1321         ctx->lc_state = LCS_INITIALIZED;
1322         ctx->lc_tags = tags;
1323         if (tags & LCT_REMEMBER) {
1324                 spin_lock(&lu_keys_guard);
1325                 list_add(&ctx->lc_remember, &lu_context_remembered);
1326                 spin_unlock(&lu_keys_guard);
1327         } else
1328                 CFS_INIT_LIST_HEAD(&ctx->lc_remember);
1329         return keys_init(ctx);
1330 }
1331 EXPORT_SYMBOL(lu_context_init);
1332
1333 /**
1334  * Finalize context data-structure. Destroy key values.
1335  */
1336 void lu_context_fini(struct lu_context *ctx)
1337 {
1338         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1339         ctx->lc_state = LCS_FINALIZED;
1340         keys_fini(ctx);
1341         spin_lock(&lu_keys_guard);
1342         list_del_init(&ctx->lc_remember);
1343         spin_unlock(&lu_keys_guard);
1344 }
1345 EXPORT_SYMBOL(lu_context_fini);
1346
1347 /**
1348  * Called before entering context.
1349  */
1350 void lu_context_enter(struct lu_context *ctx)
1351 {
1352         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1353         ctx->lc_state = LCS_ENTERED;
1354 }
1355 EXPORT_SYMBOL(lu_context_enter);
1356
1357 /**
1358  * Called after exiting from \a ctx
1359  */
1360 void lu_context_exit(struct lu_context *ctx)
1361 {
1362         int i;
1363
1364         LINVRNT(ctx->lc_state == LCS_ENTERED);
1365         ctx->lc_state = LCS_LEFT;
1366         if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
1367                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1368                         if (ctx->lc_value[i] != NULL) {
1369                                 struct lu_context_key *key;
1370
1371                                 key = lu_keys[i];
1372                                 LASSERT(key != NULL);
1373                                 if (key->lct_exit != NULL)
1374                                         key->lct_exit(ctx,
1375                                                       key, ctx->lc_value[i]);
1376                         }
1377                 }
1378         }
1379 }
1380 EXPORT_SYMBOL(lu_context_exit);
1381
1382 /**
1383  * Allocate for context all missing keys that were registered after context
1384  * creation.
1385  */
1386 int lu_context_refill(struct lu_context *ctx)
1387 {
1388         LINVRNT(ctx->lc_value != NULL);
1389         return ctx->lc_version == key_set_version ? 0 : keys_fill(ctx);
1390 }
1391 EXPORT_SYMBOL(lu_context_refill);
1392
1393 int lu_env_init(struct lu_env *env, __u32 tags)
1394 {
1395         int result;
1396
1397         env->le_ses = NULL;
1398         result = lu_context_init(&env->le_ctx, tags);
1399         if (likely(result == 0))
1400                 lu_context_enter(&env->le_ctx);
1401         return result;
1402 }
1403 EXPORT_SYMBOL(lu_env_init);
1404
1405 void lu_env_fini(struct lu_env *env)
1406 {
1407         lu_context_exit(&env->le_ctx);
1408         lu_context_fini(&env->le_ctx);
1409         env->le_ses = NULL;
1410 }
1411 EXPORT_SYMBOL(lu_env_fini);
1412
1413 int lu_env_refill(struct lu_env *env)
1414 {
1415         int result;
1416
1417         result = lu_context_refill(&env->le_ctx);
1418         if (result == 0 && env->le_ses != NULL)
1419                 result = lu_context_refill(env->le_ses);
1420         return result;
1421 }
1422 EXPORT_SYMBOL(lu_env_refill);
1423
1424 static struct shrinker *lu_site_shrinker = NULL;
1425
1426 #ifdef __KERNEL__
1427 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1428 {
1429         struct lu_site *s;
1430         struct lu_site *tmp;
1431         int cached = 0;
1432         int remain = nr;
1433         CFS_LIST_HEAD(splice);
1434
1435         if (nr != 0) {
1436                 if (!(gfp_mask & __GFP_FS))
1437                         return -1;
1438                 CDEBUG(D_INODE, "Shrink %d objects\n", nr);
1439         }
1440
1441         down(&lu_sites_guard);
1442         list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1443                 if (nr != 0) {
1444                         remain = lu_site_purge(&lu_shrink_env, s, remain);
1445                         /*
1446                          * Move just shrunk site to the tail of site list to
1447                          * assure shrinking fairness.
1448                          */
1449                         list_move_tail(&s->ls_linkage, &splice);
1450                 }
1451                 read_lock(&s->ls_guard);
1452                 cached += s->ls_total - s->ls_busy;
1453                 read_unlock(&s->ls_guard);
1454                 if (nr && remain <= 0)
1455                         break;
1456         }
1457         list_splice(&splice, lu_sites.prev);
1458         up(&lu_sites_guard);
1459
1460         cached = (cached / 100) * sysctl_vfs_cache_pressure;
1461         if (nr == 0)
1462                 CDEBUG(D_INODE, "%d objects cached\n", cached);
1463         return cached;
1464 }
1465
1466 /*
1467  * Debugging stuff.
1468  */
1469
1470 /**
1471  * Environment to be used in debugger, contains all tags.
1472  */
1473 struct lu_env lu_debugging_env;
1474
1475 /**
1476  * Debugging printer function using printk().
1477  */
1478 int lu_printk_printer(const struct lu_env *env,
1479                       void *unused, const char *format, ...)
1480 {
1481         va_list args;
1482
1483         va_start(args, format);
1484         vprintk(format, args);
1485         va_end(args);
1486         return 0;
1487 }
1488
1489 void lu_debugging_setup(void)
1490 {
1491         lu_env_init(&lu_debugging_env, ~0);
1492 }
1493
1494 void lu_context_keys_dump(void)
1495 {
1496         int i;
1497
1498         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1499                 struct lu_context_key *key;
1500
1501                 key = lu_keys[i];
1502                 if (key != NULL) {
1503                         CERROR("[%i]: %p %x (%p,%p,%p) %i %i \"%s\"@%p\n",
1504                                i, key, key->lct_tags,
1505                                key->lct_init, key->lct_fini, key->lct_exit,
1506                                key->lct_index, atomic_read(&key->lct_used),
1507                                key->lct_owner ? key->lct_owner->name : "",
1508                                key->lct_owner);
1509                         lu_ref_print(&key->lct_reference);
1510                 }
1511         }
1512 }
1513 EXPORT_SYMBOL(lu_context_keys_dump);
1514 #else  /* !__KERNEL__ */
1515 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1516 {
1517         return 0;
1518 }
1519 #endif /* __KERNEL__ */
1520
1521 int  cl_global_init(void);
1522 void cl_global_fini(void);
1523 int  lu_ref_global_init(void);
1524 void lu_ref_global_fini(void);
1525
1526 int dt_global_init(void);
1527 void dt_global_fini(void);
1528
1529 int llo_global_init(void);
1530 void llo_global_fini(void);
1531
1532 /**
1533  * Initialization of global lu_* data.
1534  */
1535 int lu_global_init(void)
1536 {
1537         int result;
1538
1539         CDEBUG(D_CONSOLE, "Lustre LU module (%p).\n", &lu_keys);
1540
1541         result = lu_ref_global_init();
1542         if (result != 0)
1543                 return result;
1544
1545         LU_CONTEXT_KEY_INIT(&lu_global_key);
1546         result = lu_context_key_register(&lu_global_key);
1547         if (result != 0)
1548                 return result;
1549         /*
1550          * At this level, we don't know what tags are needed, so allocate them
1551          * conservatively. This should not be too bad, because this
1552          * environment is global.
1553          */
1554         down(&lu_sites_guard);
1555         result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
1556         up(&lu_sites_guard);
1557         if (result != 0)
1558                 return result;
1559
1560         /*
1561          * seeks estimation: 3 seeks to read a record from oi, one to read
1562          * inode, one for ea. Unfortunately setting this high value results in
1563          * lu_object/inode cache consuming all the memory.
1564          */
1565         lu_site_shrinker = set_shrinker(DEFAULT_SEEKS, lu_cache_shrink);
1566         if (lu_site_shrinker == NULL)
1567                 return -ENOMEM;
1568
1569         result = lu_time_global_init();
1570         if (result)
1571                 GOTO(out, result);
1572
1573 #ifdef __KERNEL__
1574         result = dt_global_init();
1575         if (result)
1576                 GOTO(out, result);
1577
1578         result = llo_global_init();
1579         if (result)
1580                 GOTO(out, result);
1581 #endif
1582         result = cl_global_init();
1583 out:
1584
1585         return result;
1586 }
1587
1588 /**
1589  * Dual to lu_global_init().
1590  */
1591 void lu_global_fini(void)
1592 {
1593         cl_global_fini();
1594 #ifdef __KERNEL__
1595         llo_global_fini();
1596         dt_global_fini();
1597 #endif
1598         lu_time_global_fini();
1599         if (lu_site_shrinker != NULL) {
1600                 remove_shrinker(lu_site_shrinker);
1601                 lu_site_shrinker = NULL;
1602         }
1603
1604         lu_context_key_degister(&lu_global_key);
1605
1606         /*
1607          * Tear shrinker environment down _after_ de-registering
1608          * lu_global_key, because the latter has a value in the former.
1609          */
1610         down(&lu_sites_guard);
1611         lu_env_fini(&lu_shrink_env);
1612         up(&lu_sites_guard);
1613
1614         lu_ref_global_fini();
1615 }
1616
1617 struct lu_buf LU_BUF_NULL = {
1618         .lb_buf = NULL,
1619         .lb_len = 0
1620 };
1621 EXPORT_SYMBOL(LU_BUF_NULL);
1622
1623 /**
1624  * Output site statistical counters into a buffer. Suitable for
1625  * lprocfs_rd_*()-style functions.
1626  */
1627 int lu_site_stats_print(const struct lu_site *s, char *page, int count)
1628 {
1629         int i;
1630         int populated;
1631
1632         /*
1633          * How many hash buckets are not-empty? Don't bother with locks: it's
1634          * an estimation anyway.
1635          */
1636         for (i = 0, populated = 0; i < s->ls_hash_size; i++)
1637                 populated += !hlist_empty(&s->ls_hash[i]);
1638
1639         return snprintf(page, count, "%d %d %d/%d %d %d %d %d %d %d %d\n",
1640                         s->ls_total,
1641                         s->ls_busy,
1642                         populated,
1643                         s->ls_hash_size,
1644                         s->ls_stats.s_created,
1645                         s->ls_stats.s_cache_hit,
1646                         s->ls_stats.s_cache_miss,
1647                         s->ls_stats.s_cache_check,
1648                         s->ls_stats.s_cache_race,
1649                         s->ls_stats.s_cache_death_race,
1650                         s->ls_stats.s_lru_purged);
1651 }
1652 EXPORT_SYMBOL(lu_site_stats_print);
1653
1654 const char *lu_time_names[LU_TIME_NR] = {
1655         [LU_TIME_FIND_LOOKUP] = "find_lookup",
1656         [LU_TIME_FIND_ALLOC]  = "find_alloc",
1657         [LU_TIME_FIND_INSERT] = "find_insert"
1658 };
1659 EXPORT_SYMBOL(lu_time_names);
1660
1661 /**
1662  * Helper function to initialize a number of kmem slab caches at once.
1663  */
1664 int lu_kmem_init(struct lu_kmem_descr *caches)
1665 {
1666         int result;
1667
1668         for (result = 0; caches->ckd_cache != NULL; ++caches) {
1669                 *caches->ckd_cache = cfs_mem_cache_create(caches->ckd_name,
1670                                                           caches->ckd_size,
1671                                                           0, 0);
1672                 if (*caches->ckd_cache == NULL) {
1673                         result = -ENOMEM;
1674                         break;
1675                 }
1676         }
1677         return result;
1678 }
1679 EXPORT_SYMBOL(lu_kmem_init);
1680
1681 /**
1682  * Helper function to finalize a number of kmem slab cached at once. Dual to
1683  * lu_kmem_init().
1684  */
1685 void lu_kmem_fini(struct lu_kmem_descr *caches)
1686 {
1687         int rc;
1688
1689         for (; caches->ckd_cache != NULL; ++caches) {
1690                 if (*caches->ckd_cache != NULL) {
1691                         rc = cfs_mem_cache_destroy(*caches->ckd_cache);
1692                         LASSERTF(rc == 0, "couldn't destroy %s slab\n",
1693                                  caches->ckd_name);
1694                         *caches->ckd_cache = NULL;
1695                 }
1696         }
1697 }
1698 EXPORT_SYMBOL(lu_kmem_fini);