Whamcloud - gitweb
LU-8130 lu_object: convert lu_object cache to rhashtable
[fs/lustre-release.git] / lustre / obdclass / lu_object.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/obdclass/lu_object.c
33  *
34  * Lustre Object.
35  * These are the only exported functions, they provide some generic
36  * infrastructure for managing object devices
37  *
38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
39  */
40
41 #define DEBUG_SUBSYSTEM S_CLASS
42
43 #include <linux/delay.h>
44 #include <linux/module.h>
45 #include <linux/list.h>
46 #include <linux/processor.h>
47 #include <linux/random.h>
48
49 #include <libcfs/libcfs.h>
50 #include <libcfs/linux/linux-mem.h>
51 #include <obd_class.h>
52 #include <obd_support.h>
53 #include <lustre_disk.h>
54 #include <lustre_fid.h>
55 #include <lu_object.h>
56 #include <lu_ref.h>
57
58 struct lu_site_bkt_data {
59         /**
60          * LRU list, updated on each access to object. Protected by
61          * lsb_waitq.lock.
62          *
63          * "Cold" end of LRU is lu_site::ls_lru.next. Accessed object are
64          * moved to the lu_site::ls_lru.prev
65          */
66         struct list_head                lsb_lru;
67         /**
68          * Wait-queue signaled when an object in this site is ultimately
69          * destroyed (lu_object_free()) or initialized (lu_object_start()).
70          * It is used by lu_object_find() to wait before re-trying when
71          * object in the process of destruction is found in the hash table;
72          * or wait object to be initialized by the allocator.
73          *
74          * \see htable_lookup().
75          */
76         wait_queue_head_t               lsb_waitq;
77 };
78
79 enum {
80         LU_CACHE_PERCENT_MAX     = 50,
81         LU_CACHE_PERCENT_DEFAULT = 20
82 };
83
84 #define LU_CACHE_NR_MAX_ADJUST          512
85 #define LU_CACHE_NR_UNLIMITED           -1
86 #define LU_CACHE_NR_DEFAULT             LU_CACHE_NR_UNLIMITED
87 /** This is set to roughly (20 * OSS_NTHRS_MAX) to prevent thrashing */
88 #define LU_CACHE_NR_ZFS_LIMIT           10240
89
90 #define LU_CACHE_NR_MIN                 4096
91 #define LU_CACHE_NR_MAX                 0x80000000UL
92
93 /**
94  * Max 256 buckets, we don't want too many buckets because:
95  * - consume too much memory (currently max 16K)
96  * - avoid unbalanced LRU list
97  * With few cpus there is little gain from extra buckets, so
98  * we treat this as a maximum in lu_site_init().
99  */
100 #define LU_SITE_BKT_BITS    8
101
102 static unsigned int lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
103 module_param(lu_cache_percent, int, 0644);
104 MODULE_PARM_DESC(lu_cache_percent, "Percentage of memory to be used as lu_object cache");
105
106 static long lu_cache_nr = LU_CACHE_NR_DEFAULT;
107 module_param(lu_cache_nr, long, 0644);
108 MODULE_PARM_DESC(lu_cache_nr, "Maximum number of objects in lu_object cache");
109
110 static void lu_object_free(const struct lu_env *env, struct lu_object *o);
111 static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx);
112
113 static u32 lu_fid_hash(const void *data, u32 len, u32 seed)
114 {
115         const struct lu_fid *fid = data;
116
117         seed = cfs_hash_32(seed ^ fid->f_oid, 32);
118         seed ^= cfs_hash_64(fid->f_seq, 32);
119         return seed;
120 }
121
122 static const struct rhashtable_params obj_hash_params = {
123         .key_len        = sizeof(struct lu_fid),
124         .key_offset     = offsetof(struct lu_object_header, loh_fid),
125         .head_offset    = offsetof(struct lu_object_header, loh_hash),
126         .hashfn         = lu_fid_hash,
127         .automatic_shrinking = true,
128 };
129
130 static inline int lu_bkt_hash(struct lu_site *s, const struct lu_fid *fid)
131 {
132         return lu_fid_hash(fid, sizeof(*fid), s->ls_bkt_seed) &
133                (s->ls_bkt_cnt - 1);
134 }
135
136 wait_queue_head_t *
137 lu_site_wq_from_fid(struct lu_site *site, struct lu_fid *fid)
138 {
139         struct lu_site_bkt_data *bkt;
140
141         bkt = &site->ls_bkts[lu_bkt_hash(site, fid)];
142         return &bkt->lsb_waitq;
143 }
144 EXPORT_SYMBOL(lu_site_wq_from_fid);
145
146 /**
147  * Decrease reference counter on object. If last reference is freed, return
148  * object to the cache, unless lu_object_is_dying(o) holds. In the latter
149  * case, free object immediately.
150  */
151 void lu_object_put(const struct lu_env *env, struct lu_object *o)
152 {
153         struct lu_site_bkt_data *bkt;
154         struct lu_object_header *top = o->lo_header;
155         struct lu_site *site = o->lo_dev->ld_site;
156         struct lu_object *orig = o;
157         const struct lu_fid *fid = lu_object_fid(o);
158
159         /*
160          * till we have full fids-on-OST implemented anonymous objects
161          * are possible in OSP. such an object isn't listed in the site
162          * so we should not remove it from the site.
163          */
164         if (fid_is_zero(fid)) {
165                 LASSERT(list_empty(&top->loh_lru));
166                 if (!atomic_dec_and_test(&top->loh_ref))
167                         return;
168                 list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
169                         if (o->lo_ops->loo_object_release != NULL)
170                                 o->lo_ops->loo_object_release(env, o);
171                 }
172                 lu_object_free(env, orig);
173                 return;
174         }
175
176         bkt = &site->ls_bkts[lu_bkt_hash(site, &top->loh_fid)];
177         if (atomic_add_unless(&top->loh_ref, -1, 1)) {
178 still_active:
179                 /*
180                  * At this point the object reference is dropped and lock is
181                  * not taken, so lu_object should not be touched because it
182                  * can be freed by concurrent thread.
183                  *
184                  * Somebody may be waiting for this, currently only used for
185                  * cl_object, see cl_object_put_last().
186                  */
187                 wake_up(&bkt->lsb_waitq);
188
189                 return;
190         }
191
192         spin_lock(&bkt->lsb_waitq.lock);
193         if (!atomic_dec_and_test(&top->loh_ref)) {
194                 spin_unlock(&bkt->lsb_waitq.lock);
195                 goto still_active;
196         }
197
198         /*
199          * Refcount is zero, and cannot be incremented without taking the bkt
200          * lock, so object is stable.
201          */
202
203         /*
204          * When last reference is released, iterate over object layers, and
205          * notify them that object is no longer busy.
206          */
207         list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
208                 if (o->lo_ops->loo_object_release != NULL)
209                         o->lo_ops->loo_object_release(env, o);
210         }
211
212         /*
213          * Don't use local 'is_dying' here because if was taken without lock but
214          * here we need the latest actual value of it so check lu_object
215          * directly here.
216          */
217         if (!lu_object_is_dying(top) &&
218             (lu_object_exists(orig) || lu_object_is_cl(orig))) {
219                 LASSERT(list_empty(&top->loh_lru));
220                 list_add_tail(&top->loh_lru, &bkt->lsb_lru);
221                 spin_unlock(&bkt->lsb_waitq.lock);
222                 percpu_counter_inc(&site->ls_lru_len_counter);
223                 CDEBUG(D_INODE, "Add %p/%p to site lru. bkt: %p\n",
224                        orig, top, bkt);
225                 return;
226         }
227
228         /*
229          * If object is dying (will not be cached) then remove it from hash
230          * table (it is already not on the LRU).
231          *
232          * This is done with bucket lock held.  As the only way to acquire first
233          * reference to previously unreferenced object is through hash-table
234          * lookup (lu_object_find()) which takes the lock for first reference,
235          * no race with concurrent object lookup is possible and we can safely
236          * destroy object below.
237          */
238         if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags))
239                 rhashtable_remove_fast(&site->ls_obj_hash, &top->loh_hash,
240                                        obj_hash_params);
241
242         spin_unlock(&bkt->lsb_waitq.lock);
243         /* Object was already removed from hash above, can kill it. */
244         lu_object_free(env, orig);
245 }
246 EXPORT_SYMBOL(lu_object_put);
247
248 /**
249  * Put object and don't keep in cache. This is temporary solution for
250  * multi-site objects when its layering is not constant.
251  */
252 void lu_object_put_nocache(const struct lu_env *env, struct lu_object *o)
253 {
254         set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
255         return lu_object_put(env, o);
256 }
257 EXPORT_SYMBOL(lu_object_put_nocache);
258
259 /**
260  * Kill the object and take it out of LRU cache.
261  * Currently used by client code for layout change.
262  */
263 void lu_object_unhash(const struct lu_env *env, struct lu_object *o)
264 {
265         struct lu_object_header *top;
266
267         top = o->lo_header;
268         set_bit(LU_OBJECT_HEARD_BANSHEE, &top->loh_flags);
269         if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags)) {
270                 struct lu_site *site = o->lo_dev->ld_site;
271                 struct rhashtable *obj_hash = &site->ls_obj_hash;
272                 struct lu_site_bkt_data *bkt;
273
274                 bkt = &site->ls_bkts[lu_bkt_hash(site, &top->loh_fid)];
275                 spin_lock(&bkt->lsb_waitq.lock);
276                 if (!list_empty(&top->loh_lru)) {
277                         list_del_init(&top->loh_lru);
278                         percpu_counter_dec(&site->ls_lru_len_counter);
279                 }
280                 spin_unlock(&bkt->lsb_waitq.lock);
281
282                 rhashtable_remove_fast(obj_hash, &top->loh_hash,
283                                        obj_hash_params);
284         }
285 }
286 EXPORT_SYMBOL(lu_object_unhash);
287
288 /**
289  * Allocate new object.
290  *
291  * This follows object creation protocol, described in the comment within
292  * struct lu_device_operations definition.
293  */
294 static struct lu_object *lu_object_alloc(const struct lu_env *env,
295                                          struct lu_device *dev,
296                                          const struct lu_fid *f)
297 {
298         struct lu_object *top;
299
300         /*
301          * Create top-level object slice. This will also create
302          * lu_object_header.
303          */
304         top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
305         if (top == NULL)
306                 return ERR_PTR(-ENOMEM);
307         if (IS_ERR(top))
308                 return top;
309         /*
310          * This is the only place where object fid is assigned. It's constant
311          * after this point.
312          */
313         top->lo_header->loh_fid = *f;
314
315         return top;
316 }
317
318 /**
319  * Initialize object.
320  *
321  * This is called after object hash insertion to avoid returning an object with
322  * stale attributes.
323  */
324 static int lu_object_start(const struct lu_env *env, struct lu_device *dev,
325                            struct lu_object *top,
326                            const struct lu_object_conf *conf)
327 {
328         struct lu_object *scan;
329         struct list_head *layers;
330         unsigned int init_mask = 0;
331         unsigned int init_flag;
332         int clean;
333         int result;
334
335         layers = &top->lo_header->loh_layers;
336
337         do {
338                 /*
339                  * Call ->loo_object_init() repeatedly, until no more new
340                  * object slices are created.
341                  */
342                 clean = 1;
343                 init_flag = 1;
344                 list_for_each_entry(scan, layers, lo_linkage) {
345                         if (init_mask & init_flag)
346                                 goto next;
347                         clean = 0;
348                         scan->lo_header = top->lo_header;
349                         result = scan->lo_ops->loo_object_init(env, scan, conf);
350                         if (result)
351                                 return result;
352
353                         init_mask |= init_flag;
354 next:
355                         init_flag <<= 1;
356                 }
357         } while (!clean);
358
359         list_for_each_entry_reverse(scan, layers, lo_linkage) {
360                 if (scan->lo_ops->loo_object_start != NULL) {
361                         result = scan->lo_ops->loo_object_start(env, scan);
362                         if (result)
363                                 return result;
364                 }
365         }
366
367         lprocfs_counter_incr(dev->ld_site->ls_stats, LU_SS_CREATED);
368
369         set_bit(LU_OBJECT_INITED, &top->lo_header->loh_flags);
370
371         return 0;
372 }
373
374 /**
375  * Free an object.
376  */
377 static void lu_object_free(const struct lu_env *env, struct lu_object *o)
378 {
379         wait_queue_head_t *wq;
380         struct lu_site *site;
381         struct lu_object *scan;
382         struct list_head *layers;
383         LIST_HEAD(splice);
384
385         site = o->lo_dev->ld_site;
386         layers = &o->lo_header->loh_layers;
387         wq = lu_site_wq_from_fid(site, &o->lo_header->loh_fid);
388         /*
389          * First call ->loo_object_delete() method to release all resources.
390          */
391         list_for_each_entry_reverse(scan, layers, lo_linkage) {
392                 if (scan->lo_ops->loo_object_delete != NULL)
393                         scan->lo_ops->loo_object_delete(env, scan);
394         }
395
396         /*
397          * Then, splice object layers into stand-alone list, and call
398          * ->loo_object_free() on all layers to free memory. Splice is
399          * necessary, because lu_object_header is freed together with the
400          * top-level slice.
401          */
402         list_splice_init(layers, &splice);
403         while (!list_empty(&splice)) {
404                 /*
405                  * Free layers in bottom-to-top order, so that object header
406                  * lives as long as possible and ->loo_object_free() methods
407                  * can look at its contents.
408                  */
409                 o = container_of(splice.prev, struct lu_object, lo_linkage);
410                 list_del_init(&o->lo_linkage);
411                 LASSERT(o->lo_ops->loo_object_free != NULL);
412                 o->lo_ops->loo_object_free(env, o);
413         }
414
415         if (waitqueue_active(wq))
416                 wake_up_all(wq);
417 }
418
419 /**
420  * Free \a nr objects from the cold end of the site LRU list.
421  * if canblock is 0, then don't block awaiting for another
422  * instance of lu_site_purge() to complete
423  */
424 int lu_site_purge_objects(const struct lu_env *env, struct lu_site *s,
425                           int nr, int canblock)
426 {
427         struct lu_object_header *h;
428         struct lu_object_header *temp;
429         struct lu_site_bkt_data *bkt;
430         LIST_HEAD(dispose);
431         int                      did_sth;
432         unsigned int             start = 0;
433         int                      count;
434         int                      bnr;
435         unsigned int             i;
436
437         if (OBD_FAIL_CHECK(OBD_FAIL_OBD_NO_LRU))
438                 RETURN(0);
439
440         /*
441          * Under LRU list lock, scan LRU list and move unreferenced objects to
442          * the dispose list, removing them from LRU and hash table.
443          */
444         if (nr != ~0)
445                 start = s->ls_purge_start;
446         bnr = (nr == ~0) ? -1 : nr / s->ls_bkt_cnt + 1;
447 again:
448         /*
449          * It doesn't make any sense to make purge threads parallel, that can
450          * only bring troubles to us.  See LU-5331.
451          */
452         if (canblock != 0)
453                 mutex_lock(&s->ls_purge_mutex);
454         else if (mutex_trylock(&s->ls_purge_mutex) == 0)
455                 goto out;
456
457         did_sth = 0;
458         for (i = start; i < s->ls_bkt_cnt ; i++) {
459                 count = bnr;
460                 bkt = &s->ls_bkts[i];
461                 spin_lock(&bkt->lsb_waitq.lock);
462
463                 list_for_each_entry_safe(h, temp, &bkt->lsb_lru, loh_lru) {
464                         LASSERT(atomic_read(&h->loh_ref) == 0);
465
466                         LINVRNT(lu_bkt_hash(s, &h->loh_fid) == i);
467
468                         set_bit(LU_OBJECT_UNHASHED, &h->loh_flags);
469                         rhashtable_remove_fast(&s->ls_obj_hash, &h->loh_hash,
470                                                obj_hash_params);
471                         list_move(&h->loh_lru, &dispose);
472                         percpu_counter_dec(&s->ls_lru_len_counter);
473                         if (did_sth == 0)
474                                 did_sth = 1;
475
476                         if (nr != ~0 && --nr == 0)
477                                 break;
478
479                         if (count > 0 && --count == 0)
480                                 break;
481
482                 }
483                 spin_unlock(&bkt->lsb_waitq.lock);
484                 cond_resched();
485                 /*
486                  * Free everything on the dispose list. This is safe against
487                  * races due to the reasons described in lu_object_put().
488                  */
489                 while ((h = list_first_entry_or_null(&dispose,
490                                                      struct lu_object_header,
491                                                      loh_lru)) != NULL) {
492                         list_del_init(&h->loh_lru);
493                         lu_object_free(env, lu_object_top(h));
494                         lprocfs_counter_incr(s->ls_stats, LU_SS_LRU_PURGED);
495                 }
496
497                 if (nr == 0)
498                         break;
499         }
500         mutex_unlock(&s->ls_purge_mutex);
501
502         if (nr != 0 && did_sth && start != 0) {
503                 start = 0; /* restart from the first bucket */
504                 goto again;
505         }
506         /* race on s->ls_purge_start, but nobody cares */
507         s->ls_purge_start = i & (s->ls_bkt_cnt - 1);
508 out:
509         return nr;
510 }
511 EXPORT_SYMBOL(lu_site_purge_objects);
512
513 /*
514  * Object printing.
515  *
516  * Code below has to jump through certain loops to output object description
517  * into libcfs_debug_msg-based log. The problem is that lu_object_print()
518  * composes object description from strings that are parts of _lines_ of
519  * output (i.e., strings that are not terminated by newline). This doesn't fit
520  * very well into libcfs_debug_msg() interface that assumes that each message
521  * supplied to it is a self-contained output line.
522  *
523  * To work around this, strings are collected in a temporary buffer
524  * (implemented as a value of lu_cdebug_key key), until terminating newline
525  * character is detected.
526  *
527  */
528
529 enum {
530         /**
531          * Maximal line size.
532          *
533          * XXX overflow is not handled correctly.
534          */
535         LU_CDEBUG_LINE = 512
536 };
537
538 struct lu_cdebug_data {
539         /**
540          * Temporary buffer.
541          */
542         char lck_area[LU_CDEBUG_LINE];
543 };
544
545 /* context key constructor/destructor: lu_global_key_init, lu_global_key_fini */
546 LU_KEY_INIT_FINI(lu_global, struct lu_cdebug_data);
547
548 /**
549  * Key, holding temporary buffer. This key is registered very early by
550  * lu_global_init().
551  */
552 static struct lu_context_key lu_global_key = {
553         .lct_tags = LCT_MD_THREAD | LCT_DT_THREAD |
554                     LCT_MG_THREAD | LCT_CL_THREAD | LCT_LOCAL,
555         .lct_init = lu_global_key_init,
556         .lct_fini = lu_global_key_fini
557 };
558
559 /**
560  * Printer function emitting messages through libcfs_debug_msg().
561  */
562 int lu_cdebug_printer(const struct lu_env *env,
563                       void *cookie, const char *format, ...)
564 {
565         struct libcfs_debug_msg_data *msgdata = cookie;
566         struct lu_cdebug_data        *key;
567         int used;
568         int complete;
569         va_list args;
570
571         va_start(args, format);
572
573         key = lu_context_key_get(&env->le_ctx, &lu_global_key);
574         LASSERT(key != NULL);
575
576         used = strlen(key->lck_area);
577         complete = format[strlen(format) - 1] == '\n';
578         /*
579          * Append new chunk to the buffer.
580          */
581         vsnprintf(key->lck_area + used,
582                   ARRAY_SIZE(key->lck_area) - used, format, args);
583         if (complete) {
584                 if (cfs_cdebug_show(msgdata->msg_mask, msgdata->msg_subsys))
585                         libcfs_debug_msg(msgdata, "%s\n", key->lck_area);
586                 key->lck_area[0] = 0;
587         }
588         va_end(args);
589         return 0;
590 }
591 EXPORT_SYMBOL(lu_cdebug_printer);
592
593 /**
594  * Print object header.
595  */
596 void lu_object_header_print(const struct lu_env *env, void *cookie,
597                             lu_printer_t printer,
598                             const struct lu_object_header *hdr)
599 {
600         (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
601                    hdr, hdr->loh_flags, atomic_read(&hdr->loh_ref),
602                    PFID(&hdr->loh_fid),
603                    test_bit(LU_OBJECT_UNHASHED,
604                             &hdr->loh_flags) ? "" : " hash",
605                    list_empty(&hdr->loh_lru) ? "" : " lru",
606                    hdr->loh_attr & LOHA_EXISTS ? " exist" : "");
607 }
608 EXPORT_SYMBOL(lu_object_header_print);
609
610 /**
611  * Print human readable representation of the \a o to the \a printer.
612  */
613 void lu_object_print(const struct lu_env *env, void *cookie,
614                      lu_printer_t printer, const struct lu_object *o)
615 {
616         static const char ruler[] = "........................................";
617         struct lu_object_header *top;
618         int depth = 4;
619
620         top = o->lo_header;
621         lu_object_header_print(env, cookie, printer, top);
622         (*printer)(env, cookie, "{\n");
623
624         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
625                 /*
626                  * print `.' \a depth times followed by type name and address
627                  */
628                 (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
629                            o->lo_dev->ld_type->ldt_name, o);
630
631                 if (o->lo_ops->loo_object_print != NULL)
632                         (*o->lo_ops->loo_object_print)(env, cookie, printer, o);
633
634                 (*printer)(env, cookie, "\n");
635         }
636
637         (*printer)(env, cookie, "} header@%p\n", top);
638 }
639 EXPORT_SYMBOL(lu_object_print);
640
641 /**
642  * Check object consistency.
643  */
644 int lu_object_invariant(const struct lu_object *o)
645 {
646         struct lu_object_header *top;
647
648         top = o->lo_header;
649         list_for_each_entry(o, &top->loh_layers, lo_linkage) {
650                 if (o->lo_ops->loo_object_invariant != NULL &&
651                     !o->lo_ops->loo_object_invariant(o))
652                         return 0;
653         }
654         return 1;
655 }
656
657 /*
658  * Limit the lu_object cache to a maximum of lu_cache_nr objects.  Because the
659  * calculation for the number of objects to reclaim is not covered by a lock the
660  * maximum number of objects is capped by LU_CACHE_MAX_ADJUST.  This ensures
661  * that many concurrent threads will not accidentally purge the entire cache.
662  */
663 static void lu_object_limit(const struct lu_env *env,
664                             struct lu_device *dev)
665 {
666         u64 size, nr;
667
668         if (lu_cache_nr == LU_CACHE_NR_UNLIMITED)
669                 return;
670
671         size = atomic_read(&dev->ld_site->ls_obj_hash.nelems);
672         nr = (u64)lu_cache_nr;
673         if (size <= nr)
674                 return;
675
676         lu_site_purge_objects(env, dev->ld_site,
677                               min_t(u64, size - nr, LU_CACHE_NR_MAX_ADJUST),
678                               0);
679 }
680
681 static struct lu_object *htable_lookup(const struct lu_env *env,
682                                        struct lu_device *dev,
683                                        struct lu_site_bkt_data *bkt,
684                                        const struct lu_fid *f,
685                                        struct lu_object_header *new)
686 {
687         struct lu_site *s = dev->ld_site;
688         struct lu_object_header *h;
689
690 try_again:
691         rcu_read_lock();
692         if (new)
693                 h = rhashtable_lookup_get_insert_fast(&s->ls_obj_hash,
694                                                       &new->loh_hash,
695                                                       obj_hash_params);
696         else
697                 h = rhashtable_lookup(&s->ls_obj_hash, f, obj_hash_params);
698
699         if (IS_ERR_OR_NULL(h)) {
700                 /* Not found */
701                 if (!new)
702                         lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
703                 rcu_read_unlock();
704                 if (PTR_ERR(h) == -ENOMEM) {
705                         msleep(20);
706                         goto try_again;
707                 }
708                 lu_object_limit(env, dev);
709                 if (PTR_ERR(h) == -E2BIG)
710                         goto try_again;
711
712                 return ERR_PTR(-ENOENT);
713         }
714
715         if (atomic_inc_not_zero(&h->loh_ref)) {
716                 rcu_read_unlock();
717                 return lu_object_top(h);
718         }
719
720         spin_lock(&bkt->lsb_waitq.lock);
721         if (lu_object_is_dying(h) ||
722             test_bit(LU_OBJECT_UNHASHED, &h->loh_flags)) {
723                 spin_unlock(&bkt->lsb_waitq.lock);
724                 rcu_read_unlock();
725                 if (new) {
726                         /*
727                          * Old object might have already been removed, or will
728                          * be soon.  We need to insert our new object, so
729                          * remove the old one just in case it is still there.
730                          */
731                         rhashtable_remove_fast(&s->ls_obj_hash, &h->loh_hash,
732                                                obj_hash_params);
733                         goto try_again;
734                 }
735                 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
736                 return ERR_PTR(-ENOENT);
737         }
738         /* Now protected by spinlock */
739         rcu_read_unlock();
740
741         if (!list_empty(&h->loh_lru)) {
742                 list_del_init(&h->loh_lru);
743                 percpu_counter_dec(&s->ls_lru_len_counter);
744         }
745         atomic_inc(&h->loh_ref);
746         spin_unlock(&bkt->lsb_waitq.lock);
747         lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_HIT);
748         return lu_object_top(h);
749 }
750
751 /**
752  * Search cache for an object with the fid \a f. If such object is found,
753  * return it. Otherwise, create new object, insert it into cache and return
754  * it. In any case, additional reference is acquired on the returned object.
755  */
756 struct lu_object *lu_object_find(const struct lu_env *env,
757                                  struct lu_device *dev, const struct lu_fid *f,
758                                  const struct lu_object_conf *conf)
759 {
760         return lu_object_find_at(env, dev->ld_site->ls_top_dev, f, conf);
761 }
762 EXPORT_SYMBOL(lu_object_find);
763
764 /*
765  * Get a 'first' reference to an object that was found while looking through the
766  * hash table.
767  */
768 struct lu_object *lu_object_get_first(struct lu_object_header *h,
769                                       struct lu_device *dev)
770 {
771         struct lu_site *s = dev->ld_site;
772         struct lu_object *ret;
773
774         if (IS_ERR_OR_NULL(h) || lu_object_is_dying(h))
775                 return NULL;
776
777         ret = lu_object_locate(h, dev->ld_type);
778         if (!ret)
779                 return ret;
780
781         if (!atomic_inc_not_zero(&h->loh_ref)) {
782                 struct lu_site_bkt_data *bkt;
783
784                 bkt = &s->ls_bkts[lu_bkt_hash(s, &h->loh_fid)];
785                 spin_lock(&bkt->lsb_waitq.lock);
786                 if (!lu_object_is_dying(h) &&
787                     !test_bit(LU_OBJECT_UNHASHED, &h->loh_flags))
788                         atomic_inc(&h->loh_ref);
789                 else
790                         ret = NULL;
791                 spin_unlock(&bkt->lsb_waitq.lock);
792         }
793         return ret;
794 }
795 EXPORT_SYMBOL(lu_object_get_first);
796
797 /**
798  * Core logic of lu_object_find*() functions.
799  *
800  * Much like lu_object_find(), but top level device of object is specifically
801  * \a dev rather than top level device of the site. This interface allows
802  * objects of different "stacking" to be created within the same site.
803  */
804 struct lu_object *lu_object_find_at(const struct lu_env *env,
805                                     struct lu_device *dev,
806                                     const struct lu_fid *f,
807                                     const struct lu_object_conf *conf)
808 {
809         struct lu_object *o;
810         struct lu_object *shadow;
811         struct lu_site *s;
812         struct lu_site_bkt_data *bkt;
813         struct rhashtable *hs;
814         int rc;
815
816         ENTRY;
817
818         /*
819          * This uses standard index maintenance protocol:
820          *
821          *     - search index under lock, and return object if found;
822          *     - otherwise, unlock index, allocate new object;
823          *     - lock index and search again;
824          *     - if nothing is found (usual case), insert newly created
825          *       object into index;
826          *     - otherwise (race: other thread inserted object), free
827          *       object just allocated.
828          *     - unlock index;
829          *     - return object.
830          *
831          * For "LOC_F_NEW" case, we are sure the object is new established.
832          * It is unnecessary to perform lookup-alloc-lookup-insert, instead,
833          * just alloc and insert directly.
834          *
835          */
836         s  = dev->ld_site;
837         hs = &s->ls_obj_hash;
838
839         if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_OBD_ZERO_NLINK_RACE)))
840                 lu_site_purge(env, s, -1);
841
842         bkt = &s->ls_bkts[lu_bkt_hash(s, f)];
843         if (!(conf && conf->loc_flags & LOC_F_NEW)) {
844                 o = htable_lookup(env, dev, bkt, f, NULL);
845
846                 if (!IS_ERR(o)) {
847                         if (likely(lu_object_is_inited(o->lo_header)))
848                                 RETURN(o);
849
850                         wait_event_idle(bkt->lsb_waitq,
851                                         lu_object_is_inited(o->lo_header) ||
852                                         lu_object_is_dying(o->lo_header));
853
854                         if (lu_object_is_dying(o->lo_header)) {
855                                 lu_object_put(env, o);
856
857                                 RETURN(ERR_PTR(-ENOENT));
858                         }
859
860                         RETURN(o);
861                 }
862
863                 if (PTR_ERR(o) != -ENOENT)
864                         RETURN(o);
865         }
866
867         /*
868          * Allocate new object, NB, object is unitialized in case object
869          * is changed between allocation and hash insertion, thus the object
870          * with stale attributes is returned.
871          */
872         o = lu_object_alloc(env, dev, f);
873         if (IS_ERR(o))
874                 RETURN(o);
875
876         LASSERT(lu_fid_eq(lu_object_fid(o), f));
877
878         CFS_RACE_WAIT(OBD_FAIL_OBD_ZERO_NLINK_RACE);
879
880         if (conf && conf->loc_flags & LOC_F_NEW) {
881                 int status = rhashtable_insert_fast(hs, &o->lo_header->loh_hash,
882                                                     obj_hash_params);
883                 if (status)
884                         /* Strange error - go the slow way */
885                         shadow = htable_lookup(env, dev, bkt, f, o->lo_header);
886                 else
887                         shadow = ERR_PTR(-ENOENT);
888         } else {
889                 shadow = htable_lookup(env, dev, bkt, f, o->lo_header);
890         }
891         if (likely(PTR_ERR(shadow) == -ENOENT)) {
892                 /*
893                  * The new object has been successfully inserted.
894                  *
895                  * This may result in rather complicated operations, including
896                  * fld queries, inode loading, etc.
897                  */
898                 rc = lu_object_start(env, dev, o, conf);
899                 if (rc) {
900                         lu_object_put_nocache(env, o);
901                         RETURN(ERR_PTR(rc));
902                 }
903
904                 wake_up(&bkt->lsb_waitq);
905
906                 lu_object_limit(env, dev);
907
908                 RETURN(o);
909         }
910
911         lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_RACE);
912         lu_object_free(env, o);
913
914         if (!(conf && conf->loc_flags & LOC_F_NEW) &&
915             !IS_ERR(shadow) &&
916             !lu_object_is_inited(shadow->lo_header)) {
917                 wait_event_idle(bkt->lsb_waitq,
918                                 lu_object_is_inited(shadow->lo_header) ||
919                                 lu_object_is_dying(shadow->lo_header));
920
921                 if (lu_object_is_dying(shadow->lo_header)) {
922                         lu_object_put(env, shadow);
923
924                         RETURN(ERR_PTR(-ENOENT));
925                 }
926         }
927
928         RETURN(shadow);
929 }
930 EXPORT_SYMBOL(lu_object_find_at);
931
932 /**
933  * Find object with given fid, and return its slice belonging to given device.
934  */
935 struct lu_object *lu_object_find_slice(const struct lu_env *env,
936                                        struct lu_device *dev,
937                                        const struct lu_fid *f,
938                                        const struct lu_object_conf *conf)
939 {
940         struct lu_object *top;
941         struct lu_object *obj;
942
943         top = lu_object_find(env, dev, f, conf);
944         if (IS_ERR(top))
945                 return top;
946
947         obj = lu_object_locate(top->lo_header, dev->ld_type);
948         if (unlikely(obj == NULL)) {
949                 lu_object_put(env, top);
950                 obj = ERR_PTR(-ENOENT);
951         }
952
953         return obj;
954 }
955 EXPORT_SYMBOL(lu_object_find_slice);
956
957 int lu_device_type_init(struct lu_device_type *ldt)
958 {
959         int result = 0;
960
961         atomic_set(&ldt->ldt_device_nr, 0);
962         if (ldt->ldt_ops->ldto_init)
963                 result = ldt->ldt_ops->ldto_init(ldt);
964
965         return result;
966 }
967 EXPORT_SYMBOL(lu_device_type_init);
968
969 void lu_device_type_fini(struct lu_device_type *ldt)
970 {
971         if (ldt->ldt_ops->ldto_fini)
972                 ldt->ldt_ops->ldto_fini(ldt);
973 }
974 EXPORT_SYMBOL(lu_device_type_fini);
975
976 /**
977  * Global list of all sites on this node
978  */
979 static LIST_HEAD(lu_sites);
980 static DECLARE_RWSEM(lu_sites_guard);
981
982 /**
983  * Global environment used by site shrinker.
984  */
985 static struct lu_env lu_shrink_env;
986
987 struct lu_site_print_arg {
988         struct lu_env   *lsp_env;
989         void            *lsp_cookie;
990         lu_printer_t     lsp_printer;
991 };
992
993 static void
994 lu_site_obj_print(struct lu_object_header *h, struct lu_site_print_arg *arg)
995 {
996         if (!list_empty(&h->loh_layers)) {
997                 const struct lu_object *o;
998
999                 o = lu_object_top(h);
1000                 lu_object_print(arg->lsp_env, arg->lsp_cookie,
1001                                 arg->lsp_printer, o);
1002         } else {
1003                 lu_object_header_print(arg->lsp_env, arg->lsp_cookie,
1004                                        arg->lsp_printer, h);
1005         }
1006 }
1007
1008 /**
1009  * Print all objects in \a s.
1010  */
1011 void lu_site_print(const struct lu_env *env, struct lu_site *s, atomic_t *ref,
1012                    int msg_flag, lu_printer_t printer)
1013 {
1014         struct lu_site_print_arg arg = {
1015                 .lsp_env     = (struct lu_env *)env,
1016                 .lsp_printer = printer,
1017         };
1018         struct rhashtable_iter iter;
1019         struct lu_object_header *h;
1020         LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, msg_flag, NULL);
1021
1022         if (!s || !atomic_read(ref))
1023                 return;
1024
1025         arg.lsp_cookie = (void *)&msgdata;
1026
1027         rhashtable_walk_enter(&s->ls_obj_hash, &iter);
1028         rhashtable_walk_start(&iter);
1029         while ((h = rhashtable_walk_next(&iter)) != NULL) {
1030                 if (IS_ERR(h))
1031                         continue;
1032                 lu_site_obj_print(h, &arg);
1033         }
1034         rhashtable_walk_stop(&iter);
1035         rhashtable_walk_exit(&iter);
1036 }
1037 EXPORT_SYMBOL(lu_site_print);
1038
1039 /**
1040  * Return desired hash table order.
1041  */
1042 static void lu_htable_limits(struct lu_device *top)
1043 {
1044         unsigned long cache_size;
1045
1046         /*
1047          * For ZFS based OSDs the cache should be disabled by default.  This
1048          * allows the ZFS ARC maximum flexibility in determining what buffers
1049          * to cache.  If Lustre has objects or buffer which it wants to ensure
1050          * always stay cached it must maintain a hold on them.
1051          */
1052         if (strcmp(top->ld_type->ldt_name, LUSTRE_OSD_ZFS_NAME) == 0) {
1053                 lu_cache_nr = LU_CACHE_NR_ZFS_LIMIT;
1054                 return;
1055         }
1056
1057         /*
1058          * Calculate hash table size, assuming that we want reasonable
1059          * performance when 20% of total memory is occupied by cache of
1060          * lu_objects.
1061          *
1062          * Size of lu_object is (arbitrary) taken as 1K (together with inode).
1063          */
1064         cache_size = cfs_totalram_pages();
1065
1066 #if BITS_PER_LONG == 32
1067         /* limit hashtable size for lowmem systems to low RAM */
1068         if (cache_size > 1 << (30 - PAGE_SHIFT))
1069                 cache_size = 1 << (30 - PAGE_SHIFT) * 3 / 4;
1070 #endif
1071
1072         /* clear off unreasonable cache setting. */
1073         if (lu_cache_percent == 0 || lu_cache_percent > LU_CACHE_PERCENT_MAX) {
1074                 CWARN("obdclass: invalid lu_cache_percent: %u, it must be in the range of (0, %u]. Will use default value: %u.\n",
1075                       lu_cache_percent, LU_CACHE_PERCENT_MAX,
1076                       LU_CACHE_PERCENT_DEFAULT);
1077
1078                 lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
1079         }
1080         cache_size = cache_size / 100 * lu_cache_percent *
1081                 (PAGE_SIZE / 1024);
1082
1083         lu_cache_nr = clamp_t(typeof(cache_size), cache_size,
1084                               LU_CACHE_NR_MIN, LU_CACHE_NR_MAX);
1085 }
1086
1087 void lu_dev_add_linkage(struct lu_site *s, struct lu_device *d)
1088 {
1089         spin_lock(&s->ls_ld_lock);
1090         if (list_empty(&d->ld_linkage))
1091                 list_add(&d->ld_linkage, &s->ls_ld_linkage);
1092         spin_unlock(&s->ls_ld_lock);
1093 }
1094 EXPORT_SYMBOL(lu_dev_add_linkage);
1095
1096 void lu_dev_del_linkage(struct lu_site *s, struct lu_device *d)
1097 {
1098         spin_lock(&s->ls_ld_lock);
1099         list_del_init(&d->ld_linkage);
1100         spin_unlock(&s->ls_ld_lock);
1101 }
1102 EXPORT_SYMBOL(lu_dev_del_linkage);
1103
1104 /**
1105   * Initialize site \a s, with \a d as the top level device.
1106   */
1107 int lu_site_init(struct lu_site *s, struct lu_device *top)
1108 {
1109         struct lu_site_bkt_data *bkt;
1110         unsigned int i;
1111         int rc;
1112         ENTRY;
1113
1114         memset(s, 0, sizeof *s);
1115         mutex_init(&s->ls_purge_mutex);
1116         lu_htable_limits(top);
1117
1118 #ifdef HAVE_PERCPU_COUNTER_INIT_GFP_FLAG
1119         rc = percpu_counter_init(&s->ls_lru_len_counter, 0, GFP_NOFS);
1120 #else
1121         rc = percpu_counter_init(&s->ls_lru_len_counter, 0);
1122 #endif
1123         if (rc)
1124                 return -ENOMEM;
1125
1126         if (rhashtable_init(&s->ls_obj_hash, &obj_hash_params) != 0) {
1127                 CERROR("failed to create lu_site hash\n");
1128                 return -ENOMEM;
1129         }
1130
1131         s->ls_bkt_seed = prandom_u32();
1132         s->ls_bkt_cnt = max_t(long, 1 << LU_SITE_BKT_BITS,
1133                               2 * num_possible_cpus());
1134         s->ls_bkt_cnt = roundup_pow_of_two(s->ls_bkt_cnt);
1135         OBD_ALLOC_PTR_ARRAY_LARGE(s->ls_bkts, s->ls_bkt_cnt);
1136         if (!s->ls_bkts) {
1137                 rhashtable_destroy(&s->ls_obj_hash);
1138                 s->ls_bkts = NULL;
1139                 return -ENOMEM;
1140         }
1141
1142         for (i = 0; i < s->ls_bkt_cnt; i++) {
1143                 bkt = &s->ls_bkts[i];
1144                 INIT_LIST_HEAD(&bkt->lsb_lru);
1145                 init_waitqueue_head(&bkt->lsb_waitq);
1146         }
1147
1148         s->ls_stats = lprocfs_alloc_stats(LU_SS_LAST_STAT, 0);
1149         if (s->ls_stats == NULL) {
1150                 OBD_FREE_PTR_ARRAY_LARGE(s->ls_bkts, s->ls_bkt_cnt);
1151                 s->ls_bkts = NULL;
1152                 rhashtable_destroy(&s->ls_obj_hash);
1153                 return -ENOMEM;
1154         }
1155
1156         lprocfs_counter_init(s->ls_stats, LU_SS_CREATED,
1157                              0, "created", "created");
1158         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_HIT,
1159                              0, "cache_hit", "cache_hit");
1160         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_MISS,
1161                              0, "cache_miss", "cache_miss");
1162         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_RACE,
1163                              0, "cache_race", "cache_race");
1164         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_DEATH_RACE,
1165                              0, "cache_death_race", "cache_death_race");
1166         lprocfs_counter_init(s->ls_stats, LU_SS_LRU_PURGED,
1167                              0, "lru_purged", "lru_purged");
1168
1169         INIT_LIST_HEAD(&s->ls_linkage);
1170         s->ls_top_dev = top;
1171         top->ld_site = s;
1172         lu_device_get(top);
1173         lu_ref_add(&top->ld_reference, "site-top", s);
1174
1175         INIT_LIST_HEAD(&s->ls_ld_linkage);
1176         spin_lock_init(&s->ls_ld_lock);
1177
1178         lu_dev_add_linkage(s, top);
1179
1180         RETURN(0);
1181 }
1182 EXPORT_SYMBOL(lu_site_init);
1183
1184 /**
1185  * Finalize \a s and release its resources.
1186  */
1187 void lu_site_fini(struct lu_site *s)
1188 {
1189         down_write(&lu_sites_guard);
1190         list_del_init(&s->ls_linkage);
1191         up_write(&lu_sites_guard);
1192
1193         percpu_counter_destroy(&s->ls_lru_len_counter);
1194
1195         if (s->ls_bkts) {
1196                 rhashtable_destroy(&s->ls_obj_hash);
1197                 OBD_FREE_PTR_ARRAY_LARGE(s->ls_bkts, s->ls_bkt_cnt);
1198                 s->ls_bkts = NULL;
1199         }
1200
1201         if (s->ls_top_dev != NULL) {
1202                 s->ls_top_dev->ld_site = NULL;
1203                 lu_ref_del(&s->ls_top_dev->ld_reference, "site-top", s);
1204                 lu_device_put(s->ls_top_dev);
1205                 s->ls_top_dev = NULL;
1206         }
1207
1208         if (s->ls_stats != NULL)
1209                 lprocfs_free_stats(&s->ls_stats);
1210 }
1211 EXPORT_SYMBOL(lu_site_fini);
1212
1213 /**
1214  * Called when initialization of stack for this site is completed.
1215  */
1216 int lu_site_init_finish(struct lu_site *s)
1217 {
1218         int result;
1219         down_write(&lu_sites_guard);
1220         result = lu_context_refill(&lu_shrink_env.le_ctx);
1221         if (result == 0)
1222                 list_add(&s->ls_linkage, &lu_sites);
1223         up_write(&lu_sites_guard);
1224         return result;
1225 }
1226 EXPORT_SYMBOL(lu_site_init_finish);
1227
1228 /**
1229  * Acquire additional reference on device \a d
1230  */
1231 void lu_device_get(struct lu_device *d)
1232 {
1233         atomic_inc(&d->ld_ref);
1234 }
1235 EXPORT_SYMBOL(lu_device_get);
1236
1237 /**
1238  * Release reference on device \a d.
1239  */
1240 void lu_device_put(struct lu_device *d)
1241 {
1242         LASSERT(atomic_read(&d->ld_ref) > 0);
1243         atomic_dec(&d->ld_ref);
1244 }
1245 EXPORT_SYMBOL(lu_device_put);
1246
1247 /**
1248  * Initialize device \a d of type \a t.
1249  */
1250 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
1251 {
1252         if (atomic_inc_return(&t->ldt_device_nr) == 1 &&
1253             t->ldt_ops->ldto_start != NULL)
1254                 t->ldt_ops->ldto_start(t);
1255
1256         memset(d, 0, sizeof *d);
1257         d->ld_type = t;
1258         lu_ref_init(&d->ld_reference);
1259         INIT_LIST_HEAD(&d->ld_linkage);
1260
1261         return 0;
1262 }
1263 EXPORT_SYMBOL(lu_device_init);
1264
1265 /**
1266  * Finalize device \a d.
1267  */
1268 void lu_device_fini(struct lu_device *d)
1269 {
1270         struct lu_device_type *t = d->ld_type;
1271
1272         if (d->ld_obd != NULL) {
1273                 d->ld_obd->obd_lu_dev = NULL;
1274                 d->ld_obd = NULL;
1275         }
1276
1277         lu_ref_fini(&d->ld_reference);
1278         LASSERTF(atomic_read(&d->ld_ref) == 0,
1279                  "Refcount is %u\n", atomic_read(&d->ld_ref));
1280         LASSERT(atomic_read(&t->ldt_device_nr) > 0);
1281
1282         if (atomic_dec_and_test(&t->ldt_device_nr) &&
1283             t->ldt_ops->ldto_stop != NULL)
1284                 t->ldt_ops->ldto_stop(t);
1285 }
1286 EXPORT_SYMBOL(lu_device_fini);
1287
1288 /**
1289  * Initialize object \a o that is part of compound object \a h and was created
1290  * by device \a d.
1291  */
1292 int lu_object_init(struct lu_object *o, struct lu_object_header *h,
1293                    struct lu_device *d)
1294 {
1295         memset(o, 0, sizeof(*o));
1296         o->lo_header = h;
1297         o->lo_dev = d;
1298         lu_device_get(d);
1299         lu_ref_add_at(&d->ld_reference, &o->lo_dev_ref, "lu_object", o);
1300         INIT_LIST_HEAD(&o->lo_linkage);
1301
1302         return 0;
1303 }
1304 EXPORT_SYMBOL(lu_object_init);
1305
1306 /**
1307  * Finalize object and release its resources.
1308  */
1309 void lu_object_fini(struct lu_object *o)
1310 {
1311         struct lu_device *dev = o->lo_dev;
1312
1313         LASSERT(list_empty(&o->lo_linkage));
1314
1315         if (dev != NULL) {
1316                 lu_ref_del_at(&dev->ld_reference, &o->lo_dev_ref,
1317                               "lu_object", o);
1318                 lu_device_put(dev);
1319                 o->lo_dev = NULL;
1320         }
1321 }
1322 EXPORT_SYMBOL(lu_object_fini);
1323
1324 /**
1325  * Add object \a o as first layer of compound object \a h
1326  *
1327  * This is typically called by the ->ldo_object_alloc() method of top-level
1328  * device.
1329  */
1330 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
1331 {
1332         list_move(&o->lo_linkage, &h->loh_layers);
1333 }
1334 EXPORT_SYMBOL(lu_object_add_top);
1335
1336 /**
1337  * Add object \a o as a layer of compound object, going after \a before.
1338  *
1339  * This is typically called by the ->ldo_object_alloc() method of \a
1340  * before->lo_dev.
1341  */
1342 void lu_object_add(struct lu_object *before, struct lu_object *o)
1343 {
1344         list_move(&o->lo_linkage, &before->lo_linkage);
1345 }
1346 EXPORT_SYMBOL(lu_object_add);
1347
1348 /**
1349  * Initialize compound object.
1350  */
1351 int lu_object_header_init(struct lu_object_header *h)
1352 {
1353         memset(h, 0, sizeof *h);
1354         atomic_set(&h->loh_ref, 1);
1355         INIT_LIST_HEAD(&h->loh_lru);
1356         INIT_LIST_HEAD(&h->loh_layers);
1357         lu_ref_init(&h->loh_reference);
1358         return 0;
1359 }
1360 EXPORT_SYMBOL(lu_object_header_init);
1361
1362 /**
1363  * Finalize compound object.
1364  */
1365 void lu_object_header_fini(struct lu_object_header *h)
1366 {
1367         LASSERT(list_empty(&h->loh_layers));
1368         LASSERT(list_empty(&h->loh_lru));
1369         lu_ref_fini(&h->loh_reference);
1370 }
1371 EXPORT_SYMBOL(lu_object_header_fini);
1372
1373 /**
1374  * Given a compound object, find its slice, corresponding to the device type
1375  * \a dtype.
1376  */
1377 struct lu_object *lu_object_locate(struct lu_object_header *h,
1378                                    const struct lu_device_type *dtype)
1379 {
1380         struct lu_object *o;
1381
1382         list_for_each_entry(o, &h->loh_layers, lo_linkage) {
1383                 if (o->lo_dev->ld_type == dtype)
1384                         return o;
1385         }
1386         return NULL;
1387 }
1388 EXPORT_SYMBOL(lu_object_locate);
1389
1390 /**
1391  * Finalize and free devices in the device stack.
1392  *
1393  * Finalize device stack by purging object cache, and calling
1394  * lu_device_type_operations::ldto_device_fini() and
1395  * lu_device_type_operations::ldto_device_free() on all devices in the stack.
1396  */
1397 void lu_stack_fini(const struct lu_env *env, struct lu_device *top)
1398 {
1399         struct lu_site   *site = top->ld_site;
1400         struct lu_device *scan;
1401         struct lu_device *next;
1402
1403         lu_site_purge(env, site, ~0);
1404         for (scan = top; scan != NULL; scan = next) {
1405                 next = scan->ld_type->ldt_ops->ldto_device_fini(env, scan);
1406                 lu_ref_del(&scan->ld_reference, "lu-stack", &lu_site_init);
1407                 lu_device_put(scan);
1408         }
1409
1410         /* purge again. */
1411         lu_site_purge(env, site, ~0);
1412
1413         for (scan = top; scan != NULL; scan = next) {
1414                 const struct lu_device_type *ldt = scan->ld_type;
1415
1416                 next = ldt->ldt_ops->ldto_device_free(env, scan);
1417         }
1418 }
1419
1420 enum {
1421         /**
1422          * Maximal number of tld slots.
1423          */
1424         LU_CONTEXT_KEY_NR = 40
1425 };
1426
1427 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
1428
1429 static DECLARE_RWSEM(lu_key_initing);
1430
1431 /**
1432  * Global counter incremented whenever key is registered, unregistered,
1433  * revived or quiesced. This is used to void unnecessary calls to
1434  * lu_context_refill(). No locking is provided, as initialization and shutdown
1435  * are supposed to be externally serialized.
1436  */
1437 static atomic_t key_set_version = ATOMIC_INIT(0);
1438
1439 /**
1440  * Register new key.
1441  */
1442 int lu_context_key_register(struct lu_context_key *key)
1443 {
1444         int result;
1445         unsigned int i;
1446
1447         LASSERT(key->lct_init != NULL);
1448         LASSERT(key->lct_fini != NULL);
1449         LASSERT(key->lct_tags != 0);
1450         LASSERT(key->lct_owner != NULL);
1451
1452         result = -ENFILE;
1453         atomic_set(&key->lct_used, 1);
1454         lu_ref_init(&key->lct_reference);
1455         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1456                 if (lu_keys[i])
1457                         continue;
1458                 key->lct_index = i;
1459
1460                 if (strncmp("osd_", module_name(key->lct_owner), 4) == 0)
1461                         CFS_RACE_WAIT(OBD_FAIL_OBD_SETUP);
1462
1463                 if (cmpxchg(&lu_keys[i], NULL, key) != NULL)
1464                         continue;
1465
1466                 result = 0;
1467                 atomic_inc(&key_set_version);
1468                 break;
1469         }
1470         if (result) {
1471                 lu_ref_fini(&key->lct_reference);
1472                 atomic_set(&key->lct_used, 0);
1473         }
1474         return result;
1475 }
1476 EXPORT_SYMBOL(lu_context_key_register);
1477
1478 static void key_fini(struct lu_context *ctx, int index)
1479 {
1480         if (ctx->lc_value != NULL && ctx->lc_value[index] != NULL) {
1481                 struct lu_context_key *key;
1482
1483                 key = lu_keys[index];
1484                 LASSERT(key != NULL);
1485                 LASSERT(key->lct_fini != NULL);
1486                 LASSERT(atomic_read(&key->lct_used) > 0);
1487
1488                 key->lct_fini(ctx, key, ctx->lc_value[index]);
1489                 lu_ref_del(&key->lct_reference, "ctx", ctx);
1490                 if (atomic_dec_and_test(&key->lct_used))
1491                         wake_up_var(&key->lct_used);
1492
1493                 LASSERT(key->lct_owner != NULL);
1494                 if ((ctx->lc_tags & LCT_NOREF) == 0) {
1495                         LINVRNT(module_refcount(key->lct_owner) > 0);
1496                         module_put(key->lct_owner);
1497                 }
1498                 ctx->lc_value[index] = NULL;
1499         }
1500 }
1501
1502 /**
1503  * Deregister key.
1504  */
1505 void lu_context_key_degister(struct lu_context_key *key)
1506 {
1507         LASSERT(atomic_read(&key->lct_used) >= 1);
1508         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1509
1510         lu_context_key_quiesce(key);
1511
1512         key_fini(&lu_shrink_env.le_ctx, key->lct_index);
1513
1514         /**
1515          * Wait until all transient contexts referencing this key have
1516          * run lu_context_key::lct_fini() method.
1517          */
1518         atomic_dec(&key->lct_used);
1519         wait_var_event(&key->lct_used, atomic_read(&key->lct_used) == 0);
1520
1521         if (!WARN_ON(lu_keys[key->lct_index] == NULL))
1522                 lu_ref_fini(&key->lct_reference);
1523
1524         smp_store_release(&lu_keys[key->lct_index], NULL);
1525 }
1526 EXPORT_SYMBOL(lu_context_key_degister);
1527
1528 /**
1529  * Register a number of keys. This has to be called after all keys have been
1530  * initialized by a call to LU_CONTEXT_KEY_INIT().
1531  */
1532 int lu_context_key_register_many(struct lu_context_key *k, ...)
1533 {
1534         struct lu_context_key *key = k;
1535         va_list args;
1536         int result;
1537
1538         va_start(args, k);
1539         do {
1540                 result = lu_context_key_register(key);
1541                 if (result)
1542                         break;
1543                 key = va_arg(args, struct lu_context_key *);
1544         } while (key != NULL);
1545         va_end(args);
1546
1547         if (result != 0) {
1548                 va_start(args, k);
1549                 while (k != key) {
1550                         lu_context_key_degister(k);
1551                         k = va_arg(args, struct lu_context_key *);
1552                 }
1553                 va_end(args);
1554         }
1555
1556         return result;
1557 }
1558 EXPORT_SYMBOL(lu_context_key_register_many);
1559
1560 /**
1561  * De-register a number of keys. This is a dual to
1562  * lu_context_key_register_many().
1563  */
1564 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1565 {
1566         va_list args;
1567
1568         va_start(args, k);
1569         do {
1570                 lu_context_key_degister(k);
1571                 k = va_arg(args, struct lu_context_key*);
1572         } while (k != NULL);
1573         va_end(args);
1574 }
1575 EXPORT_SYMBOL(lu_context_key_degister_many);
1576
1577 /**
1578  * Revive a number of keys.
1579  */
1580 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1581 {
1582         va_list args;
1583
1584         va_start(args, k);
1585         do {
1586                 lu_context_key_revive(k);
1587                 k = va_arg(args, struct lu_context_key*);
1588         } while (k != NULL);
1589         va_end(args);
1590 }
1591 EXPORT_SYMBOL(lu_context_key_revive_many);
1592
1593 /**
1594  * Quiescent a number of keys.
1595  */
1596 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1597 {
1598         va_list args;
1599
1600         va_start(args, k);
1601         do {
1602                 lu_context_key_quiesce(k);
1603                 k = va_arg(args, struct lu_context_key*);
1604         } while (k != NULL);
1605         va_end(args);
1606 }
1607 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1608
1609 /**
1610  * Return value associated with key \a key in context \a ctx.
1611  */
1612 void *lu_context_key_get(const struct lu_context *ctx,
1613                          const struct lu_context_key *key)
1614 {
1615         LINVRNT(ctx->lc_state == LCS_ENTERED);
1616         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1617         LASSERT(lu_keys[key->lct_index] == key);
1618         return ctx->lc_value[key->lct_index];
1619 }
1620 EXPORT_SYMBOL(lu_context_key_get);
1621
1622 /**
1623  * List of remembered contexts. XXX document me.
1624  */
1625 static LIST_HEAD(lu_context_remembered);
1626 static DEFINE_SPINLOCK(lu_context_remembered_guard);
1627
1628 /**
1629  * Destroy \a key in all remembered contexts. This is used to destroy key
1630  * values in "shared" contexts (like service threads), when a module owning
1631  * the key is about to be unloaded.
1632  */
1633 void lu_context_key_quiesce(struct lu_context_key *key)
1634 {
1635         struct lu_context *ctx;
1636
1637         if (!(key->lct_tags & LCT_QUIESCENT)) {
1638                 /*
1639                  * The write-lock on lu_key_initing will ensure that any
1640                  * keys_fill() which didn't see LCT_QUIESCENT will have
1641                  * finished before we call key_fini().
1642                  */
1643                 down_write(&lu_key_initing);
1644                 key->lct_tags |= LCT_QUIESCENT;
1645                 up_write(&lu_key_initing);
1646
1647                 spin_lock(&lu_context_remembered_guard);
1648                 list_for_each_entry(ctx, &lu_context_remembered, lc_remember) {
1649                         spin_until_cond(READ_ONCE(ctx->lc_state) != LCS_LEAVING);
1650                         key_fini(ctx, key->lct_index);
1651                 }
1652
1653                 spin_unlock(&lu_context_remembered_guard);
1654         }
1655 }
1656
1657 void lu_context_key_revive(struct lu_context_key *key)
1658 {
1659         key->lct_tags &= ~LCT_QUIESCENT;
1660         atomic_inc(&key_set_version);
1661 }
1662
1663 static void keys_fini(struct lu_context *ctx)
1664 {
1665         unsigned int i;
1666
1667         if (ctx->lc_value == NULL)
1668                 return;
1669
1670         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1671                 key_fini(ctx, i);
1672
1673         OBD_FREE_PTR_ARRAY(ctx->lc_value, ARRAY_SIZE(lu_keys));
1674         ctx->lc_value = NULL;
1675 }
1676
1677 static int keys_fill(struct lu_context *ctx)
1678 {
1679         unsigned int i;
1680         int rc = 0;
1681
1682         /*
1683          * A serialisation with lu_context_key_quiesce() is needed, to
1684          * ensure we see LCT_QUIESCENT and don't allocate a new value
1685          * after it freed one.  The rwsem provides this.  As down_read()
1686          * does optimistic spinning while the writer is active, this is
1687          * unlikely to ever sleep.
1688          */
1689         down_read(&lu_key_initing);
1690         ctx->lc_version = atomic_read(&key_set_version);
1691
1692         LINVRNT(ctx->lc_value);
1693         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1694                 struct lu_context_key *key;
1695
1696                 key = lu_keys[i];
1697                 if (!ctx->lc_value[i] && key &&
1698                     (key->lct_tags & ctx->lc_tags) &&
1699                     /*
1700                      * Don't create values for a LCT_QUIESCENT key, as this
1701                      * will pin module owning a key.
1702                      */
1703                     !(key->lct_tags & LCT_QUIESCENT)) {
1704                         void *value;
1705
1706                         LINVRNT(key->lct_init != NULL);
1707                         LINVRNT(key->lct_index == i);
1708
1709                         LASSERT(key->lct_owner != NULL);
1710                         if (!(ctx->lc_tags & LCT_NOREF) &&
1711                             try_module_get(key->lct_owner) == 0) {
1712                                 /* module is unloading, skip this key */
1713                                 continue;
1714                         }
1715
1716                         value = key->lct_init(ctx, key);
1717                         if (unlikely(IS_ERR(value))) {
1718                                 rc = PTR_ERR(value);
1719                                 break;
1720                         }
1721
1722                         lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1723                         atomic_inc(&key->lct_used);
1724                         /*
1725                          * This is the only place in the code, where an
1726                          * element of ctx->lc_value[] array is set to non-NULL
1727                          * value.
1728                          */
1729                         ctx->lc_value[i] = value;
1730                         if (key->lct_exit != NULL)
1731                                 ctx->lc_tags |= LCT_HAS_EXIT;
1732                 }
1733         }
1734
1735         up_read(&lu_key_initing);
1736         return rc;
1737 }
1738
1739 static int keys_init(struct lu_context *ctx)
1740 {
1741         OBD_ALLOC_PTR_ARRAY(ctx->lc_value, ARRAY_SIZE(lu_keys));
1742         if (likely(ctx->lc_value != NULL))
1743                 return keys_fill(ctx);
1744
1745         return -ENOMEM;
1746 }
1747
1748 /**
1749  * Initialize context data-structure. Create values for all keys.
1750  */
1751 int lu_context_init(struct lu_context *ctx, __u32 tags)
1752 {
1753         int     rc;
1754
1755         memset(ctx, 0, sizeof *ctx);
1756         ctx->lc_state = LCS_INITIALIZED;
1757         ctx->lc_tags = tags;
1758         if (tags & LCT_REMEMBER) {
1759                 spin_lock(&lu_context_remembered_guard);
1760                 list_add(&ctx->lc_remember, &lu_context_remembered);
1761                 spin_unlock(&lu_context_remembered_guard);
1762         } else {
1763                 INIT_LIST_HEAD(&ctx->lc_remember);
1764         }
1765
1766         rc = keys_init(ctx);
1767         if (rc != 0)
1768                 lu_context_fini(ctx);
1769
1770         return rc;
1771 }
1772 EXPORT_SYMBOL(lu_context_init);
1773
1774 /**
1775  * Finalize context data-structure. Destroy key values.
1776  */
1777 void lu_context_fini(struct lu_context *ctx)
1778 {
1779         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1780         ctx->lc_state = LCS_FINALIZED;
1781
1782         if ((ctx->lc_tags & LCT_REMEMBER) == 0) {
1783                 LASSERT(list_empty(&ctx->lc_remember));
1784         } else {
1785                 /* could race with key degister */
1786                 spin_lock(&lu_context_remembered_guard);
1787                 list_del_init(&ctx->lc_remember);
1788                 spin_unlock(&lu_context_remembered_guard);
1789         }
1790         keys_fini(ctx);
1791 }
1792 EXPORT_SYMBOL(lu_context_fini);
1793
1794 /**
1795  * Called before entering context.
1796  */
1797 void lu_context_enter(struct lu_context *ctx)
1798 {
1799         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1800         ctx->lc_state = LCS_ENTERED;
1801 }
1802 EXPORT_SYMBOL(lu_context_enter);
1803
1804 /**
1805  * Called after exiting from \a ctx
1806  */
1807 void lu_context_exit(struct lu_context *ctx)
1808 {
1809         unsigned int i;
1810
1811         LINVRNT(ctx->lc_state == LCS_ENTERED);
1812         /*
1813          * Disable preempt to ensure we get a warning if
1814          * any lct_exit ever tries to sleep.  That would hurt
1815          * lu_context_key_quiesce() which spins waiting for us.
1816          * This also ensure we aren't preempted while the state
1817          * is LCS_LEAVING, as that too would cause problems for
1818          * lu_context_key_quiesce().
1819          */
1820         preempt_disable();
1821         /*
1822          * Ensure lu_context_key_quiesce() sees LCS_LEAVING
1823          * or we see LCT_QUIESCENT
1824          */
1825         smp_store_mb(ctx->lc_state, LCS_LEAVING);
1826         if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value) {
1827                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1828                         struct lu_context_key *key;
1829
1830                         key = lu_keys[i];
1831                         if (ctx->lc_value[i] &&
1832                             !(key->lct_tags & LCT_QUIESCENT) &&
1833                             key->lct_exit)
1834                                 key->lct_exit(ctx, key, ctx->lc_value[i]);
1835                 }
1836         }
1837
1838         smp_store_release(&ctx->lc_state, LCS_LEFT);
1839         preempt_enable();
1840 }
1841 EXPORT_SYMBOL(lu_context_exit);
1842
1843 /**
1844  * Allocate for context all missing keys that were registered after context
1845  * creation. key_set_version is only changed in rare cases when modules
1846  * are loaded and removed.
1847  */
1848 int lu_context_refill(struct lu_context *ctx)
1849 {
1850         if (likely(ctx->lc_version == atomic_read(&key_set_version)))
1851                 return 0;
1852
1853         return keys_fill(ctx);
1854 }
1855
1856 /**
1857  * lu_ctx_tags/lu_ses_tags will be updated if there are new types of
1858  * obd being added. Currently, this is only used on client side, specifically
1859  * for echo device client, for other stack (like ptlrpc threads), context are
1860  * predefined when the lu_device type are registered, during the module probe
1861  * phase.
1862  */
1863 u32 lu_context_tags_default = LCT_CL_THREAD;
1864 u32 lu_session_tags_default = LCT_SESSION;
1865
1866 void lu_context_tags_update(__u32 tags)
1867 {
1868         spin_lock(&lu_context_remembered_guard);
1869         lu_context_tags_default |= tags;
1870         atomic_inc(&key_set_version);
1871         spin_unlock(&lu_context_remembered_guard);
1872 }
1873 EXPORT_SYMBOL(lu_context_tags_update);
1874
1875 void lu_context_tags_clear(__u32 tags)
1876 {
1877         spin_lock(&lu_context_remembered_guard);
1878         lu_context_tags_default &= ~tags;
1879         atomic_inc(&key_set_version);
1880         spin_unlock(&lu_context_remembered_guard);
1881 }
1882 EXPORT_SYMBOL(lu_context_tags_clear);
1883
1884 void lu_session_tags_update(__u32 tags)
1885 {
1886         spin_lock(&lu_context_remembered_guard);
1887         lu_session_tags_default |= tags;
1888         atomic_inc(&key_set_version);
1889         spin_unlock(&lu_context_remembered_guard);
1890 }
1891 EXPORT_SYMBOL(lu_session_tags_update);
1892
1893 void lu_session_tags_clear(__u32 tags)
1894 {
1895         spin_lock(&lu_context_remembered_guard);
1896         lu_session_tags_default &= ~tags;
1897         atomic_inc(&key_set_version);
1898         spin_unlock(&lu_context_remembered_guard);
1899 }
1900 EXPORT_SYMBOL(lu_session_tags_clear);
1901
1902 int lu_env_init(struct lu_env *env, __u32 tags)
1903 {
1904         int result;
1905
1906         env->le_ses = NULL;
1907         result = lu_context_init(&env->le_ctx, tags);
1908         if (likely(result == 0))
1909                 lu_context_enter(&env->le_ctx);
1910         return result;
1911 }
1912 EXPORT_SYMBOL(lu_env_init);
1913
1914 void lu_env_fini(struct lu_env *env)
1915 {
1916         lu_context_exit(&env->le_ctx);
1917         lu_context_fini(&env->le_ctx);
1918         env->le_ses = NULL;
1919 }
1920 EXPORT_SYMBOL(lu_env_fini);
1921
1922 int lu_env_refill(struct lu_env *env)
1923 {
1924         int result;
1925
1926         result = lu_context_refill(&env->le_ctx);
1927         if (result == 0 && env->le_ses != NULL)
1928                 result = lu_context_refill(env->le_ses);
1929         return result;
1930 }
1931 EXPORT_SYMBOL(lu_env_refill);
1932
1933 /**
1934  * Currently, this API will only be used by echo client.
1935  * Because echo client and normal lustre client will share
1936  * same cl_env cache. So echo client needs to refresh
1937  * the env context after it get one from the cache, especially
1938  * when normal client and echo client co-exist in the same client.
1939  */
1940 int lu_env_refill_by_tags(struct lu_env *env, __u32 ctags,
1941                           __u32 stags)
1942 {
1943         int    result;
1944
1945         if ((env->le_ctx.lc_tags & ctags) != ctags) {
1946                 env->le_ctx.lc_version = 0;
1947                 env->le_ctx.lc_tags |= ctags;
1948         }
1949
1950         if (env->le_ses && (env->le_ses->lc_tags & stags) != stags) {
1951                 env->le_ses->lc_version = 0;
1952                 env->le_ses->lc_tags |= stags;
1953         }
1954
1955         result = lu_env_refill(env);
1956
1957         return result;
1958 }
1959 EXPORT_SYMBOL(lu_env_refill_by_tags);
1960
1961
1962 struct lu_env_item {
1963         struct task_struct *lei_task;   /* rhashtable key */
1964         struct rhash_head lei_linkage;
1965         struct lu_env *lei_env;
1966         struct rcu_head lei_rcu_head;
1967 };
1968
1969 static const struct rhashtable_params lu_env_rhash_params = {
1970         .key_len     = sizeof(struct task_struct *),
1971         .key_offset  = offsetof(struct lu_env_item, lei_task),
1972         .head_offset = offsetof(struct lu_env_item, lei_linkage),
1973     };
1974
1975 struct rhashtable lu_env_rhash;
1976
1977 struct lu_env_percpu {
1978         struct task_struct *lep_task;
1979         struct lu_env *lep_env ____cacheline_aligned_in_smp;
1980 };
1981
1982 static struct lu_env_percpu lu_env_percpu[NR_CPUS];
1983
1984 int lu_env_add_task(struct lu_env *env, struct task_struct *task)
1985 {
1986         struct lu_env_item *lei, *old;
1987
1988         LASSERT(env);
1989
1990         OBD_ALLOC_PTR(lei);
1991         if (!lei)
1992                 return -ENOMEM;
1993
1994         lei->lei_task = task;
1995         lei->lei_env = env;
1996
1997         old = rhashtable_lookup_get_insert_fast(&lu_env_rhash,
1998                                                 &lei->lei_linkage,
1999                                                 lu_env_rhash_params);
2000         LASSERT(!old);
2001
2002         return 0;
2003 }
2004 EXPORT_SYMBOL(lu_env_add_task);
2005
2006 int lu_env_add(struct lu_env *env)
2007 {
2008         return lu_env_add_task(env, current);
2009 }
2010 EXPORT_SYMBOL(lu_env_add);
2011
2012 static void lu_env_item_free(struct rcu_head *head)
2013 {
2014         struct lu_env_item *lei;
2015
2016         lei = container_of(head, struct lu_env_item, lei_rcu_head);
2017         OBD_FREE_PTR(lei);
2018 }
2019
2020 void lu_env_remove(struct lu_env *env)
2021 {
2022         struct lu_env_item *lei;
2023         const void *task = current;
2024         int i;
2025
2026         for_each_possible_cpu(i) {
2027                 if (lu_env_percpu[i].lep_env == env) {
2028                         LASSERT(lu_env_percpu[i].lep_task == task);
2029                         lu_env_percpu[i].lep_task = NULL;
2030                         lu_env_percpu[i].lep_env = NULL;
2031                 }
2032         }
2033
2034         /* The rcu_lock is not taking in this case since the key
2035          * used is the actual task_struct. This implies that each
2036          * object is only removed by the owning thread, so there
2037          * can never be a race on a particular object.
2038          */
2039         lei = rhashtable_lookup_fast(&lu_env_rhash, &task,
2040                                      lu_env_rhash_params);
2041         if (lei && rhashtable_remove_fast(&lu_env_rhash, &lei->lei_linkage,
2042                                           lu_env_rhash_params) == 0)
2043                 call_rcu(&lei->lei_rcu_head, lu_env_item_free);
2044 }
2045 EXPORT_SYMBOL(lu_env_remove);
2046
2047 struct lu_env *lu_env_find(void)
2048 {
2049         struct lu_env *env = NULL;
2050         struct lu_env_item *lei;
2051         const void *task = current;
2052         int i = get_cpu();
2053
2054         if (lu_env_percpu[i].lep_task == current) {
2055                 env = lu_env_percpu[i].lep_env;
2056                 put_cpu();
2057                 LASSERT(env);
2058                 return env;
2059         }
2060
2061         lei = rhashtable_lookup_fast(&lu_env_rhash, &task,
2062                                      lu_env_rhash_params);
2063         if (lei) {
2064                 env = lei->lei_env;
2065                 lu_env_percpu[i].lep_task = current;
2066                 lu_env_percpu[i].lep_env = env;
2067         }
2068         put_cpu();
2069
2070         return env;
2071 }
2072 EXPORT_SYMBOL(lu_env_find);
2073
2074 static struct shrinker *lu_site_shrinker;
2075
2076 typedef struct lu_site_stats{
2077         unsigned        lss_populated;
2078         unsigned        lss_max_search;
2079         unsigned        lss_total;
2080         unsigned        lss_busy;
2081 } lu_site_stats_t;
2082
2083 static void lu_site_stats_get(const struct lu_site *s,
2084                               lu_site_stats_t *stats)
2085 {
2086         int cnt = atomic_read(&s->ls_obj_hash.nelems);
2087         /*
2088          * percpu_counter_sum_positive() won't accept a const pointer
2089          * as it does modify the struct by taking a spinlock
2090          */
2091         struct lu_site *s2 = (struct lu_site *)s;
2092
2093         stats->lss_busy += cnt -
2094                 percpu_counter_sum_positive(&s2->ls_lru_len_counter);
2095
2096         stats->lss_total += cnt;
2097         stats->lss_max_search = 0;
2098         stats->lss_populated = 0;
2099 }
2100
2101
2102 /*
2103  * lu_cache_shrink_count() returns an approximate number of cached objects
2104  * that can be freed by shrink_slab(). A counter, which tracks the
2105  * number of items in the site's lru, is maintained in a percpu_counter
2106  * for each site. The percpu values are incremented and decremented as
2107  * objects are added or removed from the lru. The percpu values are summed
2108  * and saved whenever a percpu value exceeds a threshold. Thus the saved,
2109  * summed value at any given time may not accurately reflect the current
2110  * lru length. But this value is sufficiently accurate for the needs of
2111  * a shrinker.
2112  *
2113  * Using a per cpu counter is a compromise solution to concurrent access:
2114  * lu_object_put() can update the counter without locking the site and
2115  * lu_cache_shrink_count can sum the counters without locking each
2116  * ls_obj_hash bucket.
2117  */
2118 static unsigned long lu_cache_shrink_count(struct shrinker *sk,
2119                                            struct shrink_control *sc)
2120 {
2121         struct lu_site *s;
2122         struct lu_site *tmp;
2123         unsigned long cached = 0;
2124
2125         if (!(sc->gfp_mask & __GFP_FS))
2126                 return 0;
2127
2128         down_read(&lu_sites_guard);
2129         list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage)
2130                 cached += percpu_counter_read_positive(&s->ls_lru_len_counter);
2131         up_read(&lu_sites_guard);
2132
2133         cached = (cached / 100) * sysctl_vfs_cache_pressure;
2134         CDEBUG(D_INODE, "%ld objects cached, cache pressure %d\n",
2135                cached, sysctl_vfs_cache_pressure);
2136
2137         return cached;
2138 }
2139
2140 static unsigned long lu_cache_shrink_scan(struct shrinker *sk,
2141                                           struct shrink_control *sc)
2142 {
2143         struct lu_site *s;
2144         struct lu_site *tmp;
2145         unsigned long remain = sc->nr_to_scan;
2146         LIST_HEAD(splice);
2147
2148         if (!(sc->gfp_mask & __GFP_FS))
2149                 /* We must not take the lu_sites_guard lock when
2150                  * __GFP_FS is *not* set because of the deadlock
2151                  * possibility detailed above. Additionally,
2152                  * since we cannot determine the number of
2153                  * objects in the cache without taking this
2154                  * lock, we're in a particularly tough spot. As
2155                  * a result, we'll just lie and say our cache is
2156                  * empty. This _should_ be ok, as we can't
2157                  * reclaim objects when __GFP_FS is *not* set
2158                  * anyways.
2159                  */
2160                 return SHRINK_STOP;
2161
2162         down_write(&lu_sites_guard);
2163         list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
2164                 remain = lu_site_purge(&lu_shrink_env, s, remain);
2165                 /*
2166                  * Move just shrunk site to the tail of site list to
2167                  * assure shrinking fairness.
2168                  */
2169                 list_move_tail(&s->ls_linkage, &splice);
2170         }
2171         list_splice(&splice, lu_sites.prev);
2172         up_write(&lu_sites_guard);
2173
2174         return sc->nr_to_scan - remain;
2175 }
2176
2177 #ifndef HAVE_SHRINKER_COUNT
2178 /*
2179  * There exists a potential lock inversion deadlock scenario when using
2180  * Lustre on top of ZFS. This occurs between one of ZFS's
2181  * buf_hash_table.ht_lock's, and Lustre's lu_sites_guard lock. Essentially,
2182  * thread A will take the lu_sites_guard lock and sleep on the ht_lock,
2183  * while thread B will take the ht_lock and sleep on the lu_sites_guard
2184  * lock. Obviously neither thread will wake and drop their respective hold
2185  * on their lock.
2186  *
2187  * To prevent this from happening we must ensure the lu_sites_guard lock is
2188  * not taken while down this code path. ZFS reliably does not set the
2189  * __GFP_FS bit in its code paths, so this can be used to determine if it
2190  * is safe to take the lu_sites_guard lock.
2191  *
2192  * Ideally we should accurately return the remaining number of cached
2193  * objects without taking the lu_sites_guard lock, but this is not
2194  * possible in the current implementation.
2195  */
2196 static int lu_cache_shrink(SHRINKER_ARGS(sc, nr_to_scan, gfp_mask))
2197 {
2198         int cached = 0;
2199         struct shrink_control scv = {
2200                  .nr_to_scan = shrink_param(sc, nr_to_scan),
2201                  .gfp_mask   = shrink_param(sc, gfp_mask)
2202         };
2203
2204         CDEBUG(D_INODE, "Shrink %lu objects\n", scv.nr_to_scan);
2205
2206         if (scv.nr_to_scan != 0)
2207                 lu_cache_shrink_scan(shrinker, &scv);
2208
2209         cached = lu_cache_shrink_count(shrinker, &scv);
2210         return cached;
2211 }
2212
2213 #endif /* HAVE_SHRINKER_COUNT */
2214
2215
2216 /*
2217  * Debugging stuff.
2218  */
2219
2220 /**
2221  * Environment to be used in debugger, contains all tags.
2222  */
2223 static struct lu_env lu_debugging_env;
2224
2225 /**
2226  * Debugging printer function using printk().
2227  */
2228 int lu_printk_printer(const struct lu_env *env,
2229                       void *unused, const char *format, ...)
2230 {
2231         va_list args;
2232
2233         va_start(args, format);
2234         vprintk(format, args);
2235         va_end(args);
2236         return 0;
2237 }
2238
2239 int lu_debugging_setup(void)
2240 {
2241         return lu_env_init(&lu_debugging_env, ~0);
2242 }
2243
2244 void lu_context_keys_dump(void)
2245 {
2246         unsigned int i;
2247
2248         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
2249                 struct lu_context_key *key;
2250
2251                 key = lu_keys[i];
2252                 if (key != NULL) {
2253                         CERROR("[%d]: %p %x (%p,%p,%p) %d %d \"%s\"@%p\n",
2254                                i, key, key->lct_tags,
2255                                key->lct_init, key->lct_fini, key->lct_exit,
2256                                key->lct_index, atomic_read(&key->lct_used),
2257                                key->lct_owner ? key->lct_owner->name : "",
2258                                key->lct_owner);
2259                         lu_ref_print(&key->lct_reference);
2260                 }
2261         }
2262 }
2263
2264 /**
2265  * Initialization of global lu_* data.
2266  */
2267 int lu_global_init(void)
2268 {
2269         int result;
2270         DEF_SHRINKER_VAR(shvar, lu_cache_shrink,
2271                          lu_cache_shrink_count, lu_cache_shrink_scan);
2272
2273         CDEBUG(D_INFO, "Lustre LU module (%p).\n", &lu_keys);
2274
2275         result = lu_ref_global_init();
2276         if (result != 0)
2277                 return result;
2278
2279         LU_CONTEXT_KEY_INIT(&lu_global_key);
2280         result = lu_context_key_register(&lu_global_key);
2281         if (result != 0)
2282                 return result;
2283
2284         /*
2285          * At this level, we don't know what tags are needed, so allocate them
2286          * conservatively. This should not be too bad, because this
2287          * environment is global.
2288          */
2289         down_write(&lu_sites_guard);
2290         result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
2291         up_write(&lu_sites_guard);
2292         if (result != 0)
2293                 return result;
2294
2295         /*
2296          * seeks estimation: 3 seeks to read a record from oi, one to read
2297          * inode, one for ea. Unfortunately setting this high value results in
2298          * lu_object/inode cache consuming all the memory.
2299          */
2300         lu_site_shrinker = set_shrinker(DEFAULT_SEEKS, &shvar);
2301         if (lu_site_shrinker == NULL)
2302                 return -ENOMEM;
2303
2304         result = rhashtable_init(&lu_env_rhash, &lu_env_rhash_params);
2305
2306         return result;
2307 }
2308
2309 /**
2310  * Dual to lu_global_init().
2311  */
2312 void lu_global_fini(void)
2313 {
2314         if (lu_site_shrinker != NULL) {
2315                 remove_shrinker(lu_site_shrinker);
2316                 lu_site_shrinker = NULL;
2317         }
2318
2319         lu_context_key_degister(&lu_global_key);
2320
2321         /*
2322          * Tear shrinker environment down _after_ de-registering
2323          * lu_global_key, because the latter has a value in the former.
2324          */
2325         down_write(&lu_sites_guard);
2326         lu_env_fini(&lu_shrink_env);
2327         up_write(&lu_sites_guard);
2328
2329         rhashtable_destroy(&lu_env_rhash);
2330
2331         lu_ref_global_fini();
2332 }
2333
2334 static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx)
2335 {
2336 #ifdef CONFIG_PROC_FS
2337         struct lprocfs_counter ret;
2338
2339         lprocfs_stats_collect(stats, idx, &ret);
2340         return (__u32)ret.lc_count;
2341 #else
2342         return 0;
2343 #endif
2344 }
2345
2346 /**
2347  * Output site statistical counters into a buffer. Suitable for
2348  * lprocfs_rd_*()-style functions.
2349  */
2350 int lu_site_stats_seq_print(const struct lu_site *s, struct seq_file *m)
2351 {
2352         const struct bucket_table *tbl;
2353         lu_site_stats_t stats;
2354         unsigned int chains;
2355
2356         memset(&stats, 0, sizeof(stats));
2357         lu_site_stats_get(s, &stats);
2358
2359         rcu_read_lock();
2360         tbl = rht_dereference_rcu(s->ls_obj_hash.tbl,
2361                                   &((struct lu_site *)s)->ls_obj_hash);
2362         chains = tbl->size;
2363         rcu_read_unlock();
2364         seq_printf(m, "%d/%d %d/%u %d %d %d %d %d %d %d\n",
2365                    stats.lss_busy,
2366                    stats.lss_total,
2367                    stats.lss_populated,
2368                    chains,
2369                    stats.lss_max_search,
2370                    ls_stats_read(s->ls_stats, LU_SS_CREATED),
2371                    ls_stats_read(s->ls_stats, LU_SS_CACHE_HIT),
2372                    ls_stats_read(s->ls_stats, LU_SS_CACHE_MISS),
2373                    ls_stats_read(s->ls_stats, LU_SS_CACHE_RACE),
2374                    ls_stats_read(s->ls_stats, LU_SS_CACHE_DEATH_RACE),
2375                    ls_stats_read(s->ls_stats, LU_SS_LRU_PURGED));
2376         return 0;
2377 }
2378 EXPORT_SYMBOL(lu_site_stats_seq_print);
2379
2380 /**
2381  * Helper function to initialize a number of kmem slab caches at once.
2382  */
2383 int lu_kmem_init(struct lu_kmem_descr *caches)
2384 {
2385         int result;
2386         struct lu_kmem_descr *iter = caches;
2387
2388         for (result = 0; iter->ckd_cache != NULL; ++iter) {
2389                 *iter->ckd_cache = kmem_cache_create(iter->ckd_name,
2390                                                      iter->ckd_size,
2391                                                      0, 0, NULL);
2392                 if (*iter->ckd_cache == NULL) {
2393                         result = -ENOMEM;
2394                         /* free all previously allocated caches */
2395                         lu_kmem_fini(caches);
2396                         break;
2397                 }
2398         }
2399         return result;
2400 }
2401 EXPORT_SYMBOL(lu_kmem_init);
2402
2403 /**
2404  * Helper function to finalize a number of kmem slab cached at once. Dual to
2405  * lu_kmem_init().
2406  */
2407 void lu_kmem_fini(struct lu_kmem_descr *caches)
2408 {
2409         for (; caches->ckd_cache != NULL; ++caches) {
2410                 if (*caches->ckd_cache != NULL) {
2411                         kmem_cache_destroy(*caches->ckd_cache);
2412                         *caches->ckd_cache = NULL;
2413                 }
2414         }
2415 }
2416 EXPORT_SYMBOL(lu_kmem_fini);
2417
2418 /**
2419  * Temporary solution to be able to assign fid in ->do_create()
2420  * till we have fully-functional OST fids
2421  */
2422 void lu_object_assign_fid(const struct lu_env *env, struct lu_object *o,
2423                           const struct lu_fid *fid)
2424 {
2425         struct lu_site          *s = o->lo_dev->ld_site;
2426         struct lu_fid           *old = &o->lo_header->loh_fid;
2427         int rc;
2428
2429         LASSERT(fid_is_zero(old));
2430         *old = *fid;
2431 try_again:
2432         rc = rhashtable_lookup_insert_fast(&s->ls_obj_hash,
2433                                            &o->lo_header->loh_hash,
2434                                            obj_hash_params);
2435         /* supposed to be unique */
2436         LASSERT(rc != -EEXIST);
2437         /* handle hash table resizing */
2438         if (rc == -ENOMEM) {
2439                 msleep(20);
2440                 goto try_again;
2441         }
2442         /* trim the hash if its growing to big */
2443         lu_object_limit(env, o->lo_dev);
2444         if (rc == -E2BIG)
2445                 goto try_again;
2446
2447         LASSERTF(rc == 0, "failed hashtable insertion: rc = %d\n", rc);
2448 }
2449 EXPORT_SYMBOL(lu_object_assign_fid);
2450
2451 /**
2452  * allocates object with 0 (non-assiged) fid
2453  * XXX: temporary solution to be able to assign fid in ->do_create()
2454  *      till we have fully-functional OST fids
2455  */
2456 struct lu_object *lu_object_anon(const struct lu_env *env,
2457                                  struct lu_device *dev,
2458                                  const struct lu_object_conf *conf)
2459 {
2460         struct lu_fid fid;
2461         struct lu_object *o;
2462         int rc;
2463
2464         fid_zero(&fid);
2465         o = lu_object_alloc(env, dev, &fid);
2466         if (!IS_ERR(o)) {
2467                 rc = lu_object_start(env, dev, o, conf);
2468                 if (rc) {
2469                         lu_object_free(env, o);
2470                         return ERR_PTR(rc);
2471                 }
2472         }
2473
2474         return o;
2475 }
2476 EXPORT_SYMBOL(lu_object_anon);
2477
2478 struct lu_buf LU_BUF_NULL = {
2479         .lb_buf = NULL,
2480         .lb_len = 0
2481 };
2482 EXPORT_SYMBOL(LU_BUF_NULL);
2483
2484 void lu_buf_free(struct lu_buf *buf)
2485 {
2486         LASSERT(buf);
2487         if (buf->lb_buf) {
2488                 LASSERT(buf->lb_len > 0);
2489                 OBD_FREE_LARGE(buf->lb_buf, buf->lb_len);
2490                 buf->lb_buf = NULL;
2491                 buf->lb_len = 0;
2492         }
2493 }
2494 EXPORT_SYMBOL(lu_buf_free);
2495
2496 void lu_buf_alloc(struct lu_buf *buf, size_t size)
2497 {
2498         LASSERT(buf);
2499         LASSERT(buf->lb_buf == NULL);
2500         LASSERT(buf->lb_len == 0);
2501         OBD_ALLOC_LARGE(buf->lb_buf, size);
2502         if (likely(buf->lb_buf))
2503                 buf->lb_len = size;
2504 }
2505 EXPORT_SYMBOL(lu_buf_alloc);
2506
2507 void lu_buf_realloc(struct lu_buf *buf, size_t size)
2508 {
2509         lu_buf_free(buf);
2510         lu_buf_alloc(buf, size);
2511 }
2512 EXPORT_SYMBOL(lu_buf_realloc);
2513
2514 struct lu_buf *lu_buf_check_and_alloc(struct lu_buf *buf, size_t len)
2515 {
2516         if (buf->lb_buf == NULL && buf->lb_len == 0)
2517                 lu_buf_alloc(buf, len);
2518
2519         if ((len > buf->lb_len) && (buf->lb_buf != NULL))
2520                 lu_buf_realloc(buf, len);
2521
2522         return buf;
2523 }
2524 EXPORT_SYMBOL(lu_buf_check_and_alloc);
2525
2526 /**
2527  * Increase the size of the \a buf.
2528  * preserves old data in buffer
2529  * old buffer remains unchanged on error
2530  * \retval 0 or -ENOMEM
2531  */
2532 int lu_buf_check_and_grow(struct lu_buf *buf, size_t len)
2533 {
2534         char *ptr;
2535
2536         if (len <= buf->lb_len)
2537                 return 0;
2538
2539         OBD_ALLOC_LARGE(ptr, len);
2540         if (ptr == NULL)
2541                 return -ENOMEM;
2542
2543         /* Free the old buf */
2544         if (buf->lb_buf != NULL) {
2545                 memcpy(ptr, buf->lb_buf, buf->lb_len);
2546                 OBD_FREE_LARGE(buf->lb_buf, buf->lb_len);
2547         }
2548
2549         buf->lb_buf = ptr;
2550         buf->lb_len = len;
2551         return 0;
2552 }
2553 EXPORT_SYMBOL(lu_buf_check_and_grow);