4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
16 * You should have received a copy of the GNU General Public License
17 * version 2 along with this program; If not, see
18 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
20 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21 * CA 95054 USA or visit www.sun.com if you need additional information or
27 * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
28 * Use is subject to license terms.
30 * Copyright (c) 2011, 2012, Intel Corporation.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
36 * lustre/obdclass/lu_object.c
39 * These are the only exported functions, they provide some generic
40 * infrastructure for managing object devices
42 * Author: Nikita Danilov <nikita.danilov@sun.com>
45 #define DEBUG_SUBSYSTEM S_CLASS
47 #include "../../include/linux/libcfs/libcfs.h"
49 # include <linux/module.h>
52 #include "../../include/linux/libcfs/libcfs_hash.h"
53 #include "../include/obd_class.h"
54 #include "../include/obd_support.h"
55 #include "../include/lustre_disk.h"
56 #include "../include/lustre_fid.h"
57 #include "../include/lu_object.h"
58 #include "../include/lu_ref.h"
59 #include <linux/list.h>
61 static void lu_object_free(const struct lu_env *env, struct lu_object *o);
62 static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx);
65 * Decrease reference counter on object. If last reference is freed, return
66 * object to the cache, unless lu_object_is_dying(o) holds. In the latter
67 * case, free object immediately.
69 void lu_object_put(const struct lu_env *env, struct lu_object *o)
71 struct lu_site_bkt_data *bkt;
72 struct lu_object_header *top;
74 struct lu_object *orig;
75 struct cfs_hash_bd bd;
76 const struct lu_fid *fid;
79 site = o->lo_dev->ld_site;
83 * till we have full fids-on-OST implemented anonymous objects
84 * are possible in OSP. such an object isn't listed in the site
85 * so we should not remove it from the site.
87 fid = lu_object_fid(o);
88 if (fid_is_zero(fid)) {
89 LASSERT(top->loh_hash.next == NULL
90 && top->loh_hash.pprev == NULL);
91 LASSERT(list_empty(&top->loh_lru));
92 if (!atomic_dec_and_test(&top->loh_ref))
94 list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
95 if (o->lo_ops->loo_object_release != NULL)
96 o->lo_ops->loo_object_release(env, o);
98 lu_object_free(env, orig);
102 cfs_hash_bd_get(site->ls_obj_hash, &top->loh_fid, &bd);
103 bkt = cfs_hash_bd_extra_get(site->ls_obj_hash, &bd);
105 if (!cfs_hash_bd_dec_and_lock(site->ls_obj_hash, &bd, &top->loh_ref)) {
106 if (lu_object_is_dying(top)) {
109 * somebody may be waiting for this, currently only
110 * used for cl_object, see cl_object_put_last().
112 wake_up_all(&bkt->lsb_marche_funebre);
118 * When last reference is released, iterate over object
119 * layers, and notify them that object is no longer busy.
121 list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
122 if (o->lo_ops->loo_object_release != NULL)
123 o->lo_ops->loo_object_release(env, o);
126 if (!lu_object_is_dying(top)) {
127 LASSERT(list_empty(&top->loh_lru));
128 list_add_tail(&top->loh_lru, &bkt->lsb_lru);
130 lprocfs_counter_incr(site->ls_stats, LU_SS_LRU_LEN);
131 CDEBUG(D_INODE, "Add %p to site lru. hash: %p, bkt: %p, lru_len: %ld\n",
132 o, site->ls_obj_hash, bkt, bkt->lsb_lru_len);
133 cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
138 * If object is dying (will not be cached), removed it
139 * from hash table and LRU.
141 * This is done with hash table and LRU lists locked. As the only
142 * way to acquire first reference to previously unreferenced
143 * object is through hash-table lookup (lu_object_find()),
144 * or LRU scanning (lu_site_purge()), that are done under hash-table
145 * and LRU lock, no race with concurrent object lookup is possible
146 * and we can safely destroy object below.
148 if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags))
149 cfs_hash_bd_del_locked(site->ls_obj_hash, &bd, &top->loh_hash);
150 cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
152 * Object was already removed from hash and lru above, can
155 lu_object_free(env, orig);
157 EXPORT_SYMBOL(lu_object_put);
160 * Kill the object and take it out of LRU cache.
161 * Currently used by client code for layout change.
163 void lu_object_unhash(const struct lu_env *env, struct lu_object *o)
165 struct lu_object_header *top;
168 set_bit(LU_OBJECT_HEARD_BANSHEE, &top->loh_flags);
169 if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags)) {
170 struct lu_site *site = o->lo_dev->ld_site;
171 struct cfs_hash *obj_hash = site->ls_obj_hash;
172 struct cfs_hash_bd bd;
174 cfs_hash_bd_get_and_lock(obj_hash, &top->loh_fid, &bd, 1);
175 if (!list_empty(&top->loh_lru)) {
176 struct lu_site_bkt_data *bkt;
178 list_del_init(&top->loh_lru);
179 bkt = cfs_hash_bd_extra_get(obj_hash, &bd);
181 lprocfs_counter_decr(site->ls_stats, LU_SS_LRU_LEN);
183 cfs_hash_bd_del_locked(obj_hash, &bd, &top->loh_hash);
184 cfs_hash_bd_unlock(obj_hash, &bd, 1);
187 EXPORT_SYMBOL(lu_object_unhash);
190 * Allocate new object.
192 * This follows object creation protocol, described in the comment within
193 * struct lu_device_operations definition.
195 static struct lu_object *lu_object_alloc(const struct lu_env *env,
196 struct lu_device *dev,
197 const struct lu_fid *f,
198 const struct lu_object_conf *conf)
200 struct lu_object *scan;
201 struct lu_object *top;
202 struct list_head *layers;
203 unsigned int init_mask = 0;
204 unsigned int init_flag;
209 * Create top-level object slice. This will also create
212 top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
214 return ERR_PTR(-ENOMEM);
218 * This is the only place where object fid is assigned. It's constant
221 top->lo_header->loh_fid = *f;
222 layers = &top->lo_header->loh_layers;
226 * Call ->loo_object_init() repeatedly, until no more new
227 * object slices are created.
231 list_for_each_entry(scan, layers, lo_linkage) {
232 if (init_mask & init_flag)
235 scan->lo_header = top->lo_header;
236 result = scan->lo_ops->loo_object_init(env, scan, conf);
238 lu_object_free(env, top);
239 return ERR_PTR(result);
241 init_mask |= init_flag;
247 list_for_each_entry_reverse(scan, layers, lo_linkage) {
248 if (scan->lo_ops->loo_object_start != NULL) {
249 result = scan->lo_ops->loo_object_start(env, scan);
251 lu_object_free(env, top);
252 return ERR_PTR(result);
257 lprocfs_counter_incr(dev->ld_site->ls_stats, LU_SS_CREATED);
264 static void lu_object_free(const struct lu_env *env, struct lu_object *o)
266 struct lu_site_bkt_data *bkt;
267 struct lu_site *site;
268 struct lu_object *scan;
269 struct list_head *layers;
270 struct list_head splice;
272 site = o->lo_dev->ld_site;
273 layers = &o->lo_header->loh_layers;
274 bkt = lu_site_bkt_from_fid(site, &o->lo_header->loh_fid);
276 * First call ->loo_object_delete() method to release all resources.
278 list_for_each_entry_reverse(scan, layers, lo_linkage) {
279 if (scan->lo_ops->loo_object_delete != NULL)
280 scan->lo_ops->loo_object_delete(env, scan);
284 * Then, splice object layers into stand-alone list, and call
285 * ->loo_object_free() on all layers to free memory. Splice is
286 * necessary, because lu_object_header is freed together with the
289 INIT_LIST_HEAD(&splice);
290 list_splice_init(layers, &splice);
291 while (!list_empty(&splice)) {
293 * Free layers in bottom-to-top order, so that object header
294 * lives as long as possible and ->loo_object_free() methods
295 * can look at its contents.
297 o = container_of0(splice.prev, struct lu_object, lo_linkage);
298 list_del_init(&o->lo_linkage);
299 LASSERT(o->lo_ops->loo_object_free != NULL);
300 o->lo_ops->loo_object_free(env, o);
303 if (waitqueue_active(&bkt->lsb_marche_funebre))
304 wake_up_all(&bkt->lsb_marche_funebre);
308 * Free \a nr objects from the cold end of the site LRU list.
310 int lu_site_purge(const struct lu_env *env, struct lu_site *s, int nr)
312 struct lu_object_header *h;
313 struct lu_object_header *temp;
314 struct lu_site_bkt_data *bkt;
315 struct cfs_hash_bd bd;
316 struct cfs_hash_bd bd2;
317 struct list_head dispose;
324 if (OBD_FAIL_CHECK(OBD_FAIL_OBD_NO_LRU))
327 INIT_LIST_HEAD(&dispose);
329 * Under LRU list lock, scan LRU list and move unreferenced objects to
330 * the dispose list, removing them from LRU and hash table.
332 start = s->ls_purge_start;
333 bnr = (nr == ~0) ? -1 : nr / CFS_HASH_NBKT(s->ls_obj_hash) + 1;
336 cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
340 cfs_hash_bd_lock(s->ls_obj_hash, &bd, 1);
341 bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
343 list_for_each_entry_safe(h, temp, &bkt->lsb_lru, loh_lru) {
344 LASSERT(atomic_read(&h->loh_ref) == 0);
346 cfs_hash_bd_get(s->ls_obj_hash, &h->loh_fid, &bd2);
347 LASSERT(bd.bd_bucket == bd2.bd_bucket);
349 cfs_hash_bd_del_locked(s->ls_obj_hash,
351 list_move(&h->loh_lru, &dispose);
353 lprocfs_counter_decr(s->ls_stats, LU_SS_LRU_LEN);
357 if (nr != ~0 && --nr == 0)
360 if (count > 0 && --count == 0)
364 cfs_hash_bd_unlock(s->ls_obj_hash, &bd, 1);
367 * Free everything on the dispose list. This is safe against
368 * races due to the reasons described in lu_object_put().
370 while (!list_empty(&dispose)) {
371 h = container_of0(dispose.next,
372 struct lu_object_header, loh_lru);
373 list_del_init(&h->loh_lru);
374 lu_object_free(env, lu_object_top(h));
375 lprocfs_counter_incr(s->ls_stats, LU_SS_LRU_PURGED);
382 if (nr != 0 && did_sth && start != 0) {
383 start = 0; /* restart from the first bucket */
386 /* race on s->ls_purge_start, but nobody cares */
387 s->ls_purge_start = i % CFS_HASH_NBKT(s->ls_obj_hash);
391 EXPORT_SYMBOL(lu_site_purge);
396 * Code below has to jump through certain loops to output object description
397 * into libcfs_debug_msg-based log. The problem is that lu_object_print()
398 * composes object description from strings that are parts of _lines_ of
399 * output (i.e., strings that are not terminated by newline). This doesn't fit
400 * very well into libcfs_debug_msg() interface that assumes that each message
401 * supplied to it is a self-contained output line.
403 * To work around this, strings are collected in a temporary buffer
404 * (implemented as a value of lu_cdebug_key key), until terminating newline
405 * character is detected.
413 * XXX overflow is not handled correctly.
418 struct lu_cdebug_data {
422 char lck_area[LU_CDEBUG_LINE];
425 /* context key constructor/destructor: lu_global_key_init, lu_global_key_fini */
426 LU_KEY_INIT_FINI(lu_global, struct lu_cdebug_data);
429 * Key, holding temporary buffer. This key is registered very early by
432 static struct lu_context_key lu_global_key = {
433 .lct_tags = LCT_MD_THREAD | LCT_DT_THREAD |
434 LCT_MG_THREAD | LCT_CL_THREAD | LCT_LOCAL,
435 .lct_init = lu_global_key_init,
436 .lct_fini = lu_global_key_fini
440 * Printer function emitting messages through libcfs_debug_msg().
442 int lu_cdebug_printer(const struct lu_env *env,
443 void *cookie, const char *format, ...)
445 struct libcfs_debug_msg_data *msgdata = cookie;
446 struct lu_cdebug_data *key;
451 va_start(args, format);
453 key = lu_context_key_get(&env->le_ctx, &lu_global_key);
454 LASSERT(key != NULL);
456 used = strlen(key->lck_area);
457 complete = format[strlen(format) - 1] == '\n';
459 * Append new chunk to the buffer.
461 vsnprintf(key->lck_area + used,
462 ARRAY_SIZE(key->lck_area) - used, format, args);
464 if (cfs_cdebug_show(msgdata->msg_mask, msgdata->msg_subsys))
465 libcfs_debug_msg(msgdata, "%s", key->lck_area);
466 key->lck_area[0] = 0;
471 EXPORT_SYMBOL(lu_cdebug_printer);
474 * Print object header.
476 void lu_object_header_print(const struct lu_env *env, void *cookie,
477 lu_printer_t printer,
478 const struct lu_object_header *hdr)
480 (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
481 hdr, hdr->loh_flags, atomic_read(&hdr->loh_ref),
483 hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
484 list_empty((struct list_head *)&hdr->loh_lru) ? \
486 hdr->loh_attr & LOHA_EXISTS ? " exist":"");
488 EXPORT_SYMBOL(lu_object_header_print);
491 * Print human readable representation of the \a o to the \a printer.
493 void lu_object_print(const struct lu_env *env, void *cookie,
494 lu_printer_t printer, const struct lu_object *o)
496 static const char ruler[] = "........................................";
497 struct lu_object_header *top;
501 lu_object_header_print(env, cookie, printer, top);
502 (*printer)(env, cookie, "{\n");
504 list_for_each_entry(o, &top->loh_layers, lo_linkage) {
506 * print `.' \a depth times followed by type name and address
508 (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
509 o->lo_dev->ld_type->ldt_name, o);
511 if (o->lo_ops->loo_object_print != NULL)
512 (*o->lo_ops->loo_object_print)(env, cookie, printer, o);
514 (*printer)(env, cookie, "\n");
517 (*printer)(env, cookie, "} header@%p\n", top);
519 EXPORT_SYMBOL(lu_object_print);
521 static struct lu_object *htable_lookup(struct lu_site *s,
522 struct cfs_hash_bd *bd,
523 const struct lu_fid *f,
524 wait_queue_t *waiter,
527 struct lu_site_bkt_data *bkt;
528 struct lu_object_header *h;
529 struct hlist_node *hnode;
530 __u64 ver = cfs_hash_bd_version_get(bd);
533 return ERR_PTR(-ENOENT);
536 bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, bd);
537 /* cfs_hash_bd_peek_locked is a somehow "internal" function
538 * of cfs_hash, it doesn't add refcount on object. */
539 hnode = cfs_hash_bd_peek_locked(s->ls_obj_hash, bd, (void *)f);
541 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
542 return ERR_PTR(-ENOENT);
545 h = container_of0(hnode, struct lu_object_header, loh_hash);
546 if (likely(!lu_object_is_dying(h))) {
547 cfs_hash_get(s->ls_obj_hash, hnode);
548 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_HIT);
549 if (!list_empty(&h->loh_lru)) {
550 list_del_init(&h->loh_lru);
552 lprocfs_counter_decr(s->ls_stats, LU_SS_LRU_LEN);
554 return lu_object_top(h);
558 * Lookup found an object being destroyed this object cannot be
559 * returned (to assure that references to dying objects are eventually
560 * drained), and moreover, lookup has to wait until object is freed.
563 init_waitqueue_entry(waiter, current);
564 add_wait_queue(&bkt->lsb_marche_funebre, waiter);
565 set_current_state(TASK_UNINTERRUPTIBLE);
566 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_DEATH_RACE);
567 return ERR_PTR(-EAGAIN);
571 * Search cache for an object with the fid \a f. If such object is found,
572 * return it. Otherwise, create new object, insert it into cache and return
573 * it. In any case, additional reference is acquired on the returned object.
575 static struct lu_object *lu_object_find(const struct lu_env *env,
576 struct lu_device *dev,
577 const struct lu_fid *f,
578 const struct lu_object_conf *conf)
580 return lu_object_find_at(env, dev->ld_site->ls_top_dev, f, conf);
583 static struct lu_object *lu_object_new(const struct lu_env *env,
584 struct lu_device *dev,
585 const struct lu_fid *f,
586 const struct lu_object_conf *conf)
590 struct cfs_hash_bd bd;
592 o = lu_object_alloc(env, dev, f, conf);
596 hs = dev->ld_site->ls_obj_hash;
597 cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
598 cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
599 cfs_hash_bd_unlock(hs, &bd, 1);
604 * Core logic of lu_object_find*() functions.
606 static struct lu_object *lu_object_find_try(const struct lu_env *env,
607 struct lu_device *dev,
608 const struct lu_fid *f,
609 const struct lu_object_conf *conf,
610 wait_queue_t *waiter)
613 struct lu_object *shadow;
616 struct cfs_hash_bd bd;
620 * This uses standard index maintenance protocol:
622 * - search index under lock, and return object if found;
623 * - otherwise, unlock index, allocate new object;
624 * - lock index and search again;
625 * - if nothing is found (usual case), insert newly created
627 * - otherwise (race: other thread inserted object), free
628 * object just allocated.
632 * For "LOC_F_NEW" case, we are sure the object is new established.
633 * It is unnecessary to perform lookup-alloc-lookup-insert, instead,
634 * just alloc and insert directly.
636 * If dying object is found during index search, add @waiter to the
637 * site wait-queue and return ERR_PTR(-EAGAIN).
639 if (conf != NULL && conf->loc_flags & LOC_F_NEW)
640 return lu_object_new(env, dev, f, conf);
644 cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
645 o = htable_lookup(s, &bd, f, waiter, &version);
646 cfs_hash_bd_unlock(hs, &bd, 1);
647 if (!IS_ERR(o) || PTR_ERR(o) != -ENOENT)
651 * Allocate new object. This may result in rather complicated
652 * operations, including fld queries, inode loading, etc.
654 o = lu_object_alloc(env, dev, f, conf);
658 LASSERT(lu_fid_eq(lu_object_fid(o), f));
660 cfs_hash_bd_lock(hs, &bd, 1);
662 shadow = htable_lookup(s, &bd, f, waiter, &version);
663 if (likely(PTR_ERR(shadow) == -ENOENT)) {
664 cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
665 cfs_hash_bd_unlock(hs, &bd, 1);
669 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_RACE);
670 cfs_hash_bd_unlock(hs, &bd, 1);
671 lu_object_free(env, o);
676 * Much like lu_object_find(), but top level device of object is specifically
677 * \a dev rather than top level device of the site. This interface allows
678 * objects of different "stacking" to be created within the same site.
680 struct lu_object *lu_object_find_at(const struct lu_env *env,
681 struct lu_device *dev,
682 const struct lu_fid *f,
683 const struct lu_object_conf *conf)
685 struct lu_site_bkt_data *bkt;
686 struct lu_object *obj;
690 obj = lu_object_find_try(env, dev, f, conf, &wait);
691 if (obj != ERR_PTR(-EAGAIN))
694 * lu_object_find_try() already added waiter into the
698 bkt = lu_site_bkt_from_fid(dev->ld_site, (void *)f);
699 remove_wait_queue(&bkt->lsb_marche_funebre, &wait);
702 EXPORT_SYMBOL(lu_object_find_at);
705 * Find object with given fid, and return its slice belonging to given device.
707 struct lu_object *lu_object_find_slice(const struct lu_env *env,
708 struct lu_device *dev,
709 const struct lu_fid *f,
710 const struct lu_object_conf *conf)
712 struct lu_object *top;
713 struct lu_object *obj;
715 top = lu_object_find(env, dev, f, conf);
717 obj = lu_object_locate(top->lo_header, dev->ld_type);
719 lu_object_put(env, top);
724 EXPORT_SYMBOL(lu_object_find_slice);
727 * Global list of all device types.
729 static LIST_HEAD(lu_device_types);
731 int lu_device_type_init(struct lu_device_type *ldt)
735 INIT_LIST_HEAD(&ldt->ldt_linkage);
736 if (ldt->ldt_ops->ldto_init)
737 result = ldt->ldt_ops->ldto_init(ldt);
739 list_add(&ldt->ldt_linkage, &lu_device_types);
742 EXPORT_SYMBOL(lu_device_type_init);
744 void lu_device_type_fini(struct lu_device_type *ldt)
746 list_del_init(&ldt->ldt_linkage);
747 if (ldt->ldt_ops->ldto_fini)
748 ldt->ldt_ops->ldto_fini(ldt);
750 EXPORT_SYMBOL(lu_device_type_fini);
752 void lu_types_stop(void)
754 struct lu_device_type *ldt;
756 list_for_each_entry(ldt, &lu_device_types, ldt_linkage) {
757 if (ldt->ldt_device_nr == 0 && ldt->ldt_ops->ldto_stop)
758 ldt->ldt_ops->ldto_stop(ldt);
761 EXPORT_SYMBOL(lu_types_stop);
764 * Global list of all sites on this node
766 static LIST_HEAD(lu_sites);
767 static DEFINE_MUTEX(lu_sites_guard);
770 * Global environment used by site shrinker.
772 static struct lu_env lu_shrink_env;
774 struct lu_site_print_arg {
775 struct lu_env *lsp_env;
777 lu_printer_t lsp_printer;
781 lu_site_obj_print(struct cfs_hash *hs, struct cfs_hash_bd *bd,
782 struct hlist_node *hnode, void *data)
784 struct lu_site_print_arg *arg = (struct lu_site_print_arg *)data;
785 struct lu_object_header *h;
787 h = hlist_entry(hnode, struct lu_object_header, loh_hash);
788 if (!list_empty(&h->loh_layers)) {
789 const struct lu_object *o;
791 o = lu_object_top(h);
792 lu_object_print(arg->lsp_env, arg->lsp_cookie,
793 arg->lsp_printer, o);
795 lu_object_header_print(arg->lsp_env, arg->lsp_cookie,
796 arg->lsp_printer, h);
802 * Print all objects in \a s.
804 void lu_site_print(const struct lu_env *env, struct lu_site *s, void *cookie,
805 lu_printer_t printer)
807 struct lu_site_print_arg arg = {
808 .lsp_env = (struct lu_env *)env,
809 .lsp_cookie = cookie,
810 .lsp_printer = printer,
813 cfs_hash_for_each(s->ls_obj_hash, lu_site_obj_print, &arg);
815 EXPORT_SYMBOL(lu_site_print);
818 LU_CACHE_PERCENT_MAX = 50,
819 LU_CACHE_PERCENT_DEFAULT = 20
822 static unsigned int lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
823 module_param(lu_cache_percent, int, 0644);
824 MODULE_PARM_DESC(lu_cache_percent, "Percentage of memory to be used as lu_object cache");
827 * Return desired hash table order.
829 static int lu_htable_order(void)
831 unsigned long cache_size;
835 * Calculate hash table size, assuming that we want reasonable
836 * performance when 20% of total memory is occupied by cache of
839 * Size of lu_object is (arbitrary) taken as 1K (together with inode).
841 cache_size = totalram_pages;
843 #if BITS_PER_LONG == 32
844 /* limit hashtable size for lowmem systems to low RAM */
845 if (cache_size > 1 << (30 - PAGE_CACHE_SHIFT))
846 cache_size = 1 << (30 - PAGE_CACHE_SHIFT) * 3 / 4;
849 /* clear off unreasonable cache setting. */
850 if (lu_cache_percent == 0 || lu_cache_percent > LU_CACHE_PERCENT_MAX) {
851 CWARN("obdclass: invalid lu_cache_percent: %u, it must be in the range of (0, %u]. Will use default value: %u.\n",
852 lu_cache_percent, LU_CACHE_PERCENT_MAX,
853 LU_CACHE_PERCENT_DEFAULT);
855 lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
857 cache_size = cache_size / 100 * lu_cache_percent *
858 (PAGE_CACHE_SIZE / 1024);
860 for (bits = 1; (1 << bits) < cache_size; ++bits) {
866 static unsigned lu_obj_hop_hash(struct cfs_hash *hs,
867 const void *key, unsigned mask)
869 struct lu_fid *fid = (struct lu_fid *)key;
872 hash = fid_flatten32(fid);
873 hash += (hash >> 4) + (hash << 12); /* mixing oid and seq */
874 hash = hash_long(hash, hs->hs_bkt_bits);
876 /* give me another random factor */
877 hash -= hash_long((unsigned long)hs, fid_oid(fid) % 11 + 3);
879 hash <<= hs->hs_cur_bits - hs->hs_bkt_bits;
880 hash |= (fid_seq(fid) + fid_oid(fid)) & (CFS_HASH_NBKT(hs) - 1);
885 static void *lu_obj_hop_object(struct hlist_node *hnode)
887 return hlist_entry(hnode, struct lu_object_header, loh_hash);
890 static void *lu_obj_hop_key(struct hlist_node *hnode)
892 struct lu_object_header *h;
894 h = hlist_entry(hnode, struct lu_object_header, loh_hash);
898 static int lu_obj_hop_keycmp(const void *key, struct hlist_node *hnode)
900 struct lu_object_header *h;
902 h = hlist_entry(hnode, struct lu_object_header, loh_hash);
903 return lu_fid_eq(&h->loh_fid, (struct lu_fid *)key);
906 static void lu_obj_hop_get(struct cfs_hash *hs, struct hlist_node *hnode)
908 struct lu_object_header *h;
910 h = hlist_entry(hnode, struct lu_object_header, loh_hash);
911 atomic_inc(&h->loh_ref);
914 static void lu_obj_hop_put_locked(struct cfs_hash *hs, struct hlist_node *hnode)
916 LBUG(); /* we should never called it */
919 cfs_hash_ops_t lu_site_hash_ops = {
920 .hs_hash = lu_obj_hop_hash,
921 .hs_key = lu_obj_hop_key,
922 .hs_keycmp = lu_obj_hop_keycmp,
923 .hs_object = lu_obj_hop_object,
924 .hs_get = lu_obj_hop_get,
925 .hs_put_locked = lu_obj_hop_put_locked,
928 static void lu_dev_add_linkage(struct lu_site *s, struct lu_device *d)
930 spin_lock(&s->ls_ld_lock);
931 if (list_empty(&d->ld_linkage))
932 list_add(&d->ld_linkage, &s->ls_ld_linkage);
933 spin_unlock(&s->ls_ld_lock);
937 * Initialize site \a s, with \a d as the top level device.
939 #define LU_SITE_BITS_MIN 12
940 #define LU_SITE_BITS_MAX 24
942 * total 256 buckets, we don't want too many buckets because:
943 * - consume too much memory
944 * - avoid unbalanced LRU list
946 #define LU_SITE_BKT_BITS 8
948 int lu_site_init(struct lu_site *s, struct lu_device *top)
950 struct lu_site_bkt_data *bkt;
951 struct cfs_hash_bd bd;
956 memset(s, 0, sizeof(*s));
957 bits = lu_htable_order();
958 snprintf(name, 16, "lu_site_%s", top->ld_type->ldt_name);
959 for (bits = min(max(LU_SITE_BITS_MIN, bits), LU_SITE_BITS_MAX);
960 bits >= LU_SITE_BITS_MIN; bits--) {
961 s->ls_obj_hash = cfs_hash_create(name, bits, bits,
962 bits - LU_SITE_BKT_BITS,
965 CFS_HASH_SPIN_BKTLOCK |
966 CFS_HASH_NO_ITEMREF |
968 CFS_HASH_ASSERT_EMPTY);
969 if (s->ls_obj_hash != NULL)
973 if (s->ls_obj_hash == NULL) {
974 CERROR("failed to create lu_site hash with bits: %d\n", bits);
978 cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
979 bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
980 INIT_LIST_HEAD(&bkt->lsb_lru);
981 init_waitqueue_head(&bkt->lsb_marche_funebre);
984 s->ls_stats = lprocfs_alloc_stats(LU_SS_LAST_STAT, 0);
985 if (s->ls_stats == NULL) {
986 cfs_hash_putref(s->ls_obj_hash);
987 s->ls_obj_hash = NULL;
991 lprocfs_counter_init(s->ls_stats, LU_SS_CREATED,
992 0, "created", "created");
993 lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_HIT,
994 0, "cache_hit", "cache_hit");
995 lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_MISS,
996 0, "cache_miss", "cache_miss");
997 lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_RACE,
998 0, "cache_race", "cache_race");
999 lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_DEATH_RACE,
1000 0, "cache_death_race", "cache_death_race");
1001 lprocfs_counter_init(s->ls_stats, LU_SS_LRU_PURGED,
1002 0, "lru_purged", "lru_purged");
1004 * Unlike other counters, lru_len can be decremented so
1005 * need lc_sum instead of just lc_count
1007 lprocfs_counter_init(s->ls_stats, LU_SS_LRU_LEN,
1008 LPROCFS_CNTR_AVGMINMAX, "lru_len", "lru_len");
1010 INIT_LIST_HEAD(&s->ls_linkage);
1011 s->ls_top_dev = top;
1014 lu_ref_add(&top->ld_reference, "site-top", s);
1016 INIT_LIST_HEAD(&s->ls_ld_linkage);
1017 spin_lock_init(&s->ls_ld_lock);
1019 lu_dev_add_linkage(s, top);
1023 EXPORT_SYMBOL(lu_site_init);
1026 * Finalize \a s and release its resources.
1028 void lu_site_fini(struct lu_site *s)
1030 mutex_lock(&lu_sites_guard);
1031 list_del_init(&s->ls_linkage);
1032 mutex_unlock(&lu_sites_guard);
1034 if (s->ls_obj_hash != NULL) {
1035 cfs_hash_putref(s->ls_obj_hash);
1036 s->ls_obj_hash = NULL;
1039 if (s->ls_top_dev != NULL) {
1040 s->ls_top_dev->ld_site = NULL;
1041 lu_ref_del(&s->ls_top_dev->ld_reference, "site-top", s);
1042 lu_device_put(s->ls_top_dev);
1043 s->ls_top_dev = NULL;
1046 if (s->ls_stats != NULL)
1047 lprocfs_free_stats(&s->ls_stats);
1049 EXPORT_SYMBOL(lu_site_fini);
1052 * Called when initialization of stack for this site is completed.
1054 int lu_site_init_finish(struct lu_site *s)
1057 mutex_lock(&lu_sites_guard);
1058 result = lu_context_refill(&lu_shrink_env.le_ctx);
1060 list_add(&s->ls_linkage, &lu_sites);
1061 mutex_unlock(&lu_sites_guard);
1064 EXPORT_SYMBOL(lu_site_init_finish);
1067 * Acquire additional reference on device \a d
1069 void lu_device_get(struct lu_device *d)
1071 atomic_inc(&d->ld_ref);
1073 EXPORT_SYMBOL(lu_device_get);
1076 * Release reference on device \a d.
1078 void lu_device_put(struct lu_device *d)
1080 LASSERT(atomic_read(&d->ld_ref) > 0);
1081 atomic_dec(&d->ld_ref);
1083 EXPORT_SYMBOL(lu_device_put);
1086 * Initialize device \a d of type \a t.
1088 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
1090 if (t->ldt_device_nr++ == 0 && t->ldt_ops->ldto_start != NULL)
1091 t->ldt_ops->ldto_start(t);
1092 memset(d, 0, sizeof(*d));
1093 atomic_set(&d->ld_ref, 0);
1095 lu_ref_init(&d->ld_reference);
1096 INIT_LIST_HEAD(&d->ld_linkage);
1099 EXPORT_SYMBOL(lu_device_init);
1102 * Finalize device \a d.
1104 void lu_device_fini(struct lu_device *d)
1106 struct lu_device_type *t;
1109 if (d->ld_obd != NULL) {
1110 d->ld_obd->obd_lu_dev = NULL;
1114 lu_ref_fini(&d->ld_reference);
1115 LASSERTF(atomic_read(&d->ld_ref) == 0,
1116 "Refcount is %u\n", atomic_read(&d->ld_ref));
1117 LASSERT(t->ldt_device_nr > 0);
1118 if (--t->ldt_device_nr == 0 && t->ldt_ops->ldto_stop != NULL)
1119 t->ldt_ops->ldto_stop(t);
1121 EXPORT_SYMBOL(lu_device_fini);
1124 * Initialize object \a o that is part of compound object \a h and was created
1127 int lu_object_init(struct lu_object *o, struct lu_object_header *h,
1128 struct lu_device *d)
1130 memset(o, 0, sizeof(*o));
1134 lu_ref_add_at(&d->ld_reference, &o->lo_dev_ref, "lu_object", o);
1135 INIT_LIST_HEAD(&o->lo_linkage);
1139 EXPORT_SYMBOL(lu_object_init);
1142 * Finalize object and release its resources.
1144 void lu_object_fini(struct lu_object *o)
1146 struct lu_device *dev = o->lo_dev;
1148 LASSERT(list_empty(&o->lo_linkage));
1151 lu_ref_del_at(&dev->ld_reference, &o->lo_dev_ref,
1157 EXPORT_SYMBOL(lu_object_fini);
1160 * Add object \a o as first layer of compound object \a h
1162 * This is typically called by the ->ldo_object_alloc() method of top-level
1165 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
1167 list_move(&o->lo_linkage, &h->loh_layers);
1169 EXPORT_SYMBOL(lu_object_add_top);
1172 * Add object \a o as a layer of compound object, going after \a before.
1174 * This is typically called by the ->ldo_object_alloc() method of \a
1177 void lu_object_add(struct lu_object *before, struct lu_object *o)
1179 list_move(&o->lo_linkage, &before->lo_linkage);
1181 EXPORT_SYMBOL(lu_object_add);
1184 * Initialize compound object.
1186 int lu_object_header_init(struct lu_object_header *h)
1188 memset(h, 0, sizeof(*h));
1189 atomic_set(&h->loh_ref, 1);
1190 INIT_HLIST_NODE(&h->loh_hash);
1191 INIT_LIST_HEAD(&h->loh_lru);
1192 INIT_LIST_HEAD(&h->loh_layers);
1193 lu_ref_init(&h->loh_reference);
1196 EXPORT_SYMBOL(lu_object_header_init);
1199 * Finalize compound object.
1201 void lu_object_header_fini(struct lu_object_header *h)
1203 LASSERT(list_empty(&h->loh_layers));
1204 LASSERT(list_empty(&h->loh_lru));
1205 LASSERT(hlist_unhashed(&h->loh_hash));
1206 lu_ref_fini(&h->loh_reference);
1208 EXPORT_SYMBOL(lu_object_header_fini);
1211 * Given a compound object, find its slice, corresponding to the device type
1214 struct lu_object *lu_object_locate(struct lu_object_header *h,
1215 const struct lu_device_type *dtype)
1217 struct lu_object *o;
1219 list_for_each_entry(o, &h->loh_layers, lo_linkage) {
1220 if (o->lo_dev->ld_type == dtype)
1225 EXPORT_SYMBOL(lu_object_locate);
1230 * Finalize and free devices in the device stack.
1232 * Finalize device stack by purging object cache, and calling
1233 * lu_device_type_operations::ldto_device_fini() and
1234 * lu_device_type_operations::ldto_device_free() on all devices in the stack.
1236 void lu_stack_fini(const struct lu_env *env, struct lu_device *top)
1238 struct lu_site *site = top->ld_site;
1239 struct lu_device *scan;
1240 struct lu_device *next;
1242 lu_site_purge(env, site, ~0);
1243 for (scan = top; scan != NULL; scan = next) {
1244 next = scan->ld_type->ldt_ops->ldto_device_fini(env, scan);
1245 lu_ref_del(&scan->ld_reference, "lu-stack", &lu_site_init);
1246 lu_device_put(scan);
1250 lu_site_purge(env, site, ~0);
1252 for (scan = top; scan != NULL; scan = next) {
1253 const struct lu_device_type *ldt = scan->ld_type;
1254 struct obd_type *type;
1256 next = ldt->ldt_ops->ldto_device_free(env, scan);
1257 type = ldt->ldt_obd_type;
1260 class_put_type(type);
1264 EXPORT_SYMBOL(lu_stack_fini);
1268 * Maximal number of tld slots.
1270 LU_CONTEXT_KEY_NR = 40
1273 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
1275 static DEFINE_SPINLOCK(lu_keys_guard);
1278 * Global counter incremented whenever key is registered, unregistered,
1279 * revived or quiesced. This is used to void unnecessary calls to
1280 * lu_context_refill(). No locking is provided, as initialization and shutdown
1281 * are supposed to be externally serialized.
1283 static unsigned key_set_version;
1288 int lu_context_key_register(struct lu_context_key *key)
1293 LASSERT(key->lct_init != NULL);
1294 LASSERT(key->lct_fini != NULL);
1295 LASSERT(key->lct_tags != 0);
1298 spin_lock(&lu_keys_guard);
1299 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1300 if (lu_keys[i] == NULL) {
1302 atomic_set(&key->lct_used, 1);
1304 lu_ref_init(&key->lct_reference);
1310 spin_unlock(&lu_keys_guard);
1313 EXPORT_SYMBOL(lu_context_key_register);
1315 static void key_fini(struct lu_context *ctx, int index)
1317 if (ctx->lc_value != NULL && ctx->lc_value[index] != NULL) {
1318 struct lu_context_key *key;
1320 key = lu_keys[index];
1321 LASSERT(key != NULL);
1322 LASSERT(key->lct_fini != NULL);
1323 LASSERT(atomic_read(&key->lct_used) > 1);
1325 key->lct_fini(ctx, key, ctx->lc_value[index]);
1326 lu_ref_del(&key->lct_reference, "ctx", ctx);
1327 atomic_dec(&key->lct_used);
1329 if ((ctx->lc_tags & LCT_NOREF) == 0) {
1330 #ifdef CONFIG_MODULE_UNLOAD
1331 LINVRNT(module_refcount(key->lct_owner) > 0);
1333 module_put(key->lct_owner);
1335 ctx->lc_value[index] = NULL;
1342 void lu_context_key_degister(struct lu_context_key *key)
1344 LASSERT(atomic_read(&key->lct_used) >= 1);
1345 LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1347 lu_context_key_quiesce(key);
1350 spin_lock(&lu_keys_guard);
1351 key_fini(&lu_shrink_env.le_ctx, key->lct_index);
1352 if (lu_keys[key->lct_index]) {
1353 lu_keys[key->lct_index] = NULL;
1354 lu_ref_fini(&key->lct_reference);
1356 spin_unlock(&lu_keys_guard);
1358 LASSERTF(atomic_read(&key->lct_used) == 1,
1359 "key has instances: %d\n",
1360 atomic_read(&key->lct_used));
1362 EXPORT_SYMBOL(lu_context_key_degister);
1365 * Register a number of keys. This has to be called after all keys have been
1366 * initialized by a call to LU_CONTEXT_KEY_INIT().
1368 int lu_context_key_register_many(struct lu_context_key *k, ...)
1370 struct lu_context_key *key = k;
1376 result = lu_context_key_register(key);
1379 key = va_arg(args, struct lu_context_key *);
1380 } while (key != NULL);
1386 lu_context_key_degister(k);
1387 k = va_arg(args, struct lu_context_key *);
1394 EXPORT_SYMBOL(lu_context_key_register_many);
1397 * De-register a number of keys. This is a dual to
1398 * lu_context_key_register_many().
1400 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1406 lu_context_key_degister(k);
1407 k = va_arg(args, struct lu_context_key*);
1408 } while (k != NULL);
1411 EXPORT_SYMBOL(lu_context_key_degister_many);
1414 * Revive a number of keys.
1416 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1422 lu_context_key_revive(k);
1423 k = va_arg(args, struct lu_context_key*);
1424 } while (k != NULL);
1427 EXPORT_SYMBOL(lu_context_key_revive_many);
1430 * Quiescent a number of keys.
1432 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1438 lu_context_key_quiesce(k);
1439 k = va_arg(args, struct lu_context_key*);
1440 } while (k != NULL);
1443 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1446 * Return value associated with key \a key in context \a ctx.
1448 void *lu_context_key_get(const struct lu_context *ctx,
1449 const struct lu_context_key *key)
1451 LINVRNT(ctx->lc_state == LCS_ENTERED);
1452 LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1453 LASSERT(lu_keys[key->lct_index] == key);
1454 return ctx->lc_value[key->lct_index];
1456 EXPORT_SYMBOL(lu_context_key_get);
1459 * List of remembered contexts. XXX document me.
1461 static LIST_HEAD(lu_context_remembered);
1464 * Destroy \a key in all remembered contexts. This is used to destroy key
1465 * values in "shared" contexts (like service threads), when a module owning
1466 * the key is about to be unloaded.
1468 void lu_context_key_quiesce(struct lu_context_key *key)
1470 struct lu_context *ctx;
1472 if (!(key->lct_tags & LCT_QUIESCENT)) {
1474 * XXX layering violation.
1476 key->lct_tags |= LCT_QUIESCENT;
1478 * XXX memory barrier has to go here.
1480 spin_lock(&lu_keys_guard);
1481 list_for_each_entry(ctx, &lu_context_remembered,
1483 key_fini(ctx, key->lct_index);
1484 spin_unlock(&lu_keys_guard);
1488 EXPORT_SYMBOL(lu_context_key_quiesce);
1490 void lu_context_key_revive(struct lu_context_key *key)
1492 key->lct_tags &= ~LCT_QUIESCENT;
1495 EXPORT_SYMBOL(lu_context_key_revive);
1497 static void keys_fini(struct lu_context *ctx)
1501 if (ctx->lc_value == NULL)
1504 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1507 kfree(ctx->lc_value);
1508 ctx->lc_value = NULL;
1511 static int keys_fill(struct lu_context *ctx)
1515 LINVRNT(ctx->lc_value != NULL);
1516 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1517 struct lu_context_key *key;
1520 if (ctx->lc_value[i] == NULL && key != NULL &&
1521 (key->lct_tags & ctx->lc_tags) &&
1523 * Don't create values for a LCT_QUIESCENT key, as this
1524 * will pin module owning a key.
1526 !(key->lct_tags & LCT_QUIESCENT)) {
1529 LINVRNT(key->lct_init != NULL);
1530 LINVRNT(key->lct_index == i);
1532 value = key->lct_init(ctx, key);
1534 return PTR_ERR(value);
1536 if (!(ctx->lc_tags & LCT_NOREF))
1537 try_module_get(key->lct_owner);
1538 lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1539 atomic_inc(&key->lct_used);
1541 * This is the only place in the code, where an
1542 * element of ctx->lc_value[] array is set to non-NULL
1545 ctx->lc_value[i] = value;
1546 if (key->lct_exit != NULL)
1547 ctx->lc_tags |= LCT_HAS_EXIT;
1549 ctx->lc_version = key_set_version;
1554 static int keys_init(struct lu_context *ctx)
1556 ctx->lc_value = kcalloc(ARRAY_SIZE(lu_keys), sizeof(ctx->lc_value[0]),
1558 if (likely(ctx->lc_value != NULL))
1559 return keys_fill(ctx);
1565 * Initialize context data-structure. Create values for all keys.
1567 int lu_context_init(struct lu_context *ctx, __u32 tags)
1571 memset(ctx, 0, sizeof(*ctx));
1572 ctx->lc_state = LCS_INITIALIZED;
1573 ctx->lc_tags = tags;
1574 if (tags & LCT_REMEMBER) {
1575 spin_lock(&lu_keys_guard);
1576 list_add(&ctx->lc_remember, &lu_context_remembered);
1577 spin_unlock(&lu_keys_guard);
1579 INIT_LIST_HEAD(&ctx->lc_remember);
1582 rc = keys_init(ctx);
1584 lu_context_fini(ctx);
1588 EXPORT_SYMBOL(lu_context_init);
1591 * Finalize context data-structure. Destroy key values.
1593 void lu_context_fini(struct lu_context *ctx)
1595 LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1596 ctx->lc_state = LCS_FINALIZED;
1598 if ((ctx->lc_tags & LCT_REMEMBER) == 0) {
1599 LASSERT(list_empty(&ctx->lc_remember));
1602 } else { /* could race with key degister */
1603 spin_lock(&lu_keys_guard);
1605 list_del_init(&ctx->lc_remember);
1606 spin_unlock(&lu_keys_guard);
1609 EXPORT_SYMBOL(lu_context_fini);
1612 * Called before entering context.
1614 void lu_context_enter(struct lu_context *ctx)
1616 LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1617 ctx->lc_state = LCS_ENTERED;
1619 EXPORT_SYMBOL(lu_context_enter);
1622 * Called after exiting from \a ctx
1624 void lu_context_exit(struct lu_context *ctx)
1628 LINVRNT(ctx->lc_state == LCS_ENTERED);
1629 ctx->lc_state = LCS_LEFT;
1630 if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
1631 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1632 if (ctx->lc_value[i] != NULL) {
1633 struct lu_context_key *key;
1636 LASSERT(key != NULL);
1637 if (key->lct_exit != NULL)
1639 key, ctx->lc_value[i]);
1644 EXPORT_SYMBOL(lu_context_exit);
1647 * Allocate for context all missing keys that were registered after context
1648 * creation. key_set_version is only changed in rare cases when modules
1649 * are loaded and removed.
1651 int lu_context_refill(struct lu_context *ctx)
1653 return likely(ctx->lc_version == key_set_version) ? 0 : keys_fill(ctx);
1655 EXPORT_SYMBOL(lu_context_refill);
1658 * lu_ctx_tags/lu_ses_tags will be updated if there are new types of
1659 * obd being added. Currently, this is only used on client side, specifically
1660 * for echo device client, for other stack (like ptlrpc threads), context are
1661 * predefined when the lu_device type are registered, during the module probe
1664 __u32 lu_context_tags_default;
1665 __u32 lu_session_tags_default;
1667 int lu_env_init(struct lu_env *env, __u32 tags)
1672 result = lu_context_init(&env->le_ctx, tags);
1673 if (likely(result == 0))
1674 lu_context_enter(&env->le_ctx);
1677 EXPORT_SYMBOL(lu_env_init);
1679 void lu_env_fini(struct lu_env *env)
1681 lu_context_exit(&env->le_ctx);
1682 lu_context_fini(&env->le_ctx);
1685 EXPORT_SYMBOL(lu_env_fini);
1687 int lu_env_refill(struct lu_env *env)
1691 result = lu_context_refill(&env->le_ctx);
1692 if (result == 0 && env->le_ses != NULL)
1693 result = lu_context_refill(env->le_ses);
1696 EXPORT_SYMBOL(lu_env_refill);
1698 struct lu_site_stats {
1699 unsigned lss_populated;
1700 unsigned lss_max_search;
1705 static void lu_site_stats_get(struct cfs_hash *hs,
1706 struct lu_site_stats *stats, int populated)
1708 struct cfs_hash_bd bd;
1711 cfs_hash_for_each_bucket(hs, &bd, i) {
1712 struct lu_site_bkt_data *bkt = cfs_hash_bd_extra_get(hs, &bd);
1713 struct hlist_head *hhead;
1715 cfs_hash_bd_lock(hs, &bd, 1);
1717 cfs_hash_bd_count_get(&bd) - bkt->lsb_lru_len;
1718 stats->lss_total += cfs_hash_bd_count_get(&bd);
1719 stats->lss_max_search = max((int)stats->lss_max_search,
1720 cfs_hash_bd_depmax_get(&bd));
1722 cfs_hash_bd_unlock(hs, &bd, 1);
1726 cfs_hash_bd_for_each_hlist(hs, &bd, hhead) {
1727 if (!hlist_empty(hhead))
1728 stats->lss_populated++;
1730 cfs_hash_bd_unlock(hs, &bd, 1);
1736 * lu_cache_shrink_count returns the number of cached objects that are
1737 * candidates to be freed by shrink_slab(). A counter, which tracks
1738 * the number of items in the site's lru, is maintained in the per cpu
1739 * stats of each site. The counter is incremented when an object is added
1740 * to a site's lru and decremented when one is removed. The number of
1741 * free-able objects is the sum of all per cpu counters for all sites.
1743 * Using a per cpu counter is a compromise solution to concurrent access:
1744 * lu_object_put() can update the counter without locking the site and
1745 * lu_cache_shrink_count can sum the counters without locking each
1746 * ls_obj_hash bucket.
1748 static unsigned long lu_cache_shrink_count(struct shrinker *sk,
1749 struct shrink_control *sc)
1752 struct lu_site *tmp;
1753 unsigned long cached = 0;
1755 if (!(sc->gfp_mask & __GFP_FS))
1758 mutex_lock(&lu_sites_guard);
1759 list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1760 cached += ls_stats_read(s->ls_stats, LU_SS_LRU_LEN);
1762 mutex_unlock(&lu_sites_guard);
1764 cached = (cached / 100) * sysctl_vfs_cache_pressure;
1765 CDEBUG(D_INODE, "%ld objects cached, cache pressure %d\n",
1766 cached, sysctl_vfs_cache_pressure);
1771 static unsigned long lu_cache_shrink_scan(struct shrinker *sk,
1772 struct shrink_control *sc)
1775 struct lu_site *tmp;
1776 unsigned long remain = sc->nr_to_scan, freed = 0;
1779 if (!(sc->gfp_mask & __GFP_FS))
1780 /* We must not take the lu_sites_guard lock when
1781 * __GFP_FS is *not* set because of the deadlock
1782 * possibility detailed above. Additionally,
1783 * since we cannot determine the number of
1784 * objects in the cache without taking this
1785 * lock, we're in a particularly tough spot. As
1786 * a result, we'll just lie and say our cache is
1787 * empty. This _should_ be ok, as we can't
1788 * reclaim objects when __GFP_FS is *not* set
1793 mutex_lock(&lu_sites_guard);
1794 list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1795 freed = lu_site_purge(&lu_shrink_env, s, remain);
1798 * Move just shrunk site to the tail of site list to
1799 * assure shrinking fairness.
1801 list_move_tail(&s->ls_linkage, &splice);
1803 list_splice(&splice, lu_sites.prev);
1804 mutex_unlock(&lu_sites_guard);
1806 return sc->nr_to_scan - remain;
1810 * Debugging printer function using printk().
1812 static struct shrinker lu_site_shrinker = {
1813 .count_objects = lu_cache_shrink_count,
1814 .scan_objects = lu_cache_shrink_scan,
1815 .seeks = DEFAULT_SEEKS,
1819 * Initialization of global lu_* data.
1821 int lu_global_init(void)
1825 CDEBUG(D_INFO, "Lustre LU module (%p).\n", &lu_keys);
1827 result = lu_ref_global_init();
1831 LU_CONTEXT_KEY_INIT(&lu_global_key);
1832 result = lu_context_key_register(&lu_global_key);
1837 * At this level, we don't know what tags are needed, so allocate them
1838 * conservatively. This should not be too bad, because this
1839 * environment is global.
1841 mutex_lock(&lu_sites_guard);
1842 result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
1843 mutex_unlock(&lu_sites_guard);
1848 * seeks estimation: 3 seeks to read a record from oi, one to read
1849 * inode, one for ea. Unfortunately setting this high value results in
1850 * lu_object/inode cache consuming all the memory.
1852 register_shrinker(&lu_site_shrinker);
1858 * Dual to lu_global_init().
1860 void lu_global_fini(void)
1862 unregister_shrinker(&lu_site_shrinker);
1863 lu_context_key_degister(&lu_global_key);
1866 * Tear shrinker environment down _after_ de-registering
1867 * lu_global_key, because the latter has a value in the former.
1869 mutex_lock(&lu_sites_guard);
1870 lu_env_fini(&lu_shrink_env);
1871 mutex_unlock(&lu_sites_guard);
1873 lu_ref_global_fini();
1876 static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx)
1878 struct lprocfs_counter ret;
1880 lprocfs_stats_collect(stats, idx, &ret);
1881 if (idx == LU_SS_LRU_LEN)
1883 * protect against counter on cpu A being decremented
1884 * before counter is incremented on cpu B; unlikely
1886 return (__u32)((ret.lc_sum > 0) ? ret.lc_sum : 0);
1888 return (__u32)ret.lc_count;
1892 * Output site statistical counters into a buffer. Suitable for
1893 * lprocfs_rd_*()-style functions.
1895 int lu_site_stats_print(const struct lu_site *s, struct seq_file *m)
1897 struct lu_site_stats stats;
1899 memset(&stats, 0, sizeof(stats));
1900 lu_site_stats_get(s->ls_obj_hash, &stats, 1);
1902 seq_printf(m, "%d/%d %d/%d %d %d %d %d %d %d %d %d\n",
1905 stats.lss_populated,
1906 CFS_HASH_NHLIST(s->ls_obj_hash),
1907 stats.lss_max_search,
1908 ls_stats_read(s->ls_stats, LU_SS_CREATED),
1909 ls_stats_read(s->ls_stats, LU_SS_CACHE_HIT),
1910 ls_stats_read(s->ls_stats, LU_SS_CACHE_MISS),
1911 ls_stats_read(s->ls_stats, LU_SS_CACHE_RACE),
1912 ls_stats_read(s->ls_stats, LU_SS_CACHE_DEATH_RACE),
1913 ls_stats_read(s->ls_stats, LU_SS_LRU_PURGED),
1914 ls_stats_read(s->ls_stats, LU_SS_LRU_LEN));
1917 EXPORT_SYMBOL(lu_site_stats_print);
1920 * Helper function to initialize a number of kmem slab caches at once.
1922 int lu_kmem_init(struct lu_kmem_descr *caches)
1925 struct lu_kmem_descr *iter = caches;
1927 for (result = 0; iter->ckd_cache != NULL; ++iter) {
1928 *iter->ckd_cache = kmem_cache_create(iter->ckd_name,
1931 if (*iter->ckd_cache == NULL) {
1933 /* free all previously allocated caches */
1934 lu_kmem_fini(caches);
1940 EXPORT_SYMBOL(lu_kmem_init);
1943 * Helper function to finalize a number of kmem slab cached at once. Dual to
1946 void lu_kmem_fini(struct lu_kmem_descr *caches)
1948 for (; caches->ckd_cache != NULL; ++caches) {
1949 kmem_cache_destroy(*caches->ckd_cache);
1950 *caches->ckd_cache = NULL;
1953 EXPORT_SYMBOL(lu_kmem_fini);