-/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
- * vim:expandtab:shiftwidth=8:tabstop=8:
- *
+/*
* GPL HEADER START
*
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
/*
* Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
+ *
+ * Copyright (c) 2011, 2014, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
*/
#define DEBUG_SUBSYSTEM S_CLASS
-#ifndef EXPORT_SYMTAB
-# define EXPORT_SYMTAB
-#endif
#include <libcfs/libcfs.h>
-
-#ifdef __KERNEL__
-# include <linux/module.h>
-#endif
-
-/* hash_long() */
-#include <libcfs/libcfs_hash.h>
+#include <linux/module.h>
+#include <libcfs/libcfs_hash.h> /* hash_long() */
#include <obd_class.h>
#include <obd_support.h>
#include <lustre_disk.h>
#include <lustre_fid.h>
#include <lu_object.h>
+#include <lu_ref.h>
#include <libcfs/list.h>
-/* lu_time_global_{init,fini}() */
-#include <lu_time.h>
+
+enum {
+ LU_CACHE_PERCENT_MAX = 50,
+ LU_CACHE_PERCENT_DEFAULT = 20
+};
+
+#define LU_CACHE_NR_MAX_ADJUST 128
+#define LU_CACHE_NR_UNLIMITED -1
+#define LU_CACHE_NR_DEFAULT LU_CACHE_NR_UNLIMITED
+#define LU_CACHE_NR_LDISKFS_LIMIT LU_CACHE_NR_UNLIMITED
+/** This is set to roughly (20 * OSS_NTHRS_MAX) to prevent thrashing */
+#define LU_CACHE_NR_ZFS_LIMIT 10240
+
+#define LU_SITE_BITS_MIN 12
+#define LU_SITE_BITS_MAX 24
+/**
+ * total 256 buckets, we don't want too many buckets because:
+ * - consume too much memory
+ * - avoid unbalanced LRU list
+ */
+#define LU_SITE_BKT_BITS 8
+
+
+static unsigned int lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
+CFS_MODULE_PARM(lu_cache_percent, "i", int, 0644,
+ "Percentage of memory to be used as lu_object cache");
+
+static long lu_cache_nr = LU_CACHE_NR_DEFAULT;
+CFS_MODULE_PARM(lu_cache_nr, "l", long, 0644,
+ "Maximum number of objects in lu_object cache");
static void lu_object_free(const struct lu_env *env, struct lu_object *o);
struct lu_site *site;
struct lu_object *orig;
cfs_hash_bd_t bd;
+ const struct lu_fid *fid;
top = o->lo_header;
site = o->lo_dev->ld_site;
orig = o;
+ /*
+ * till we have full fids-on-OST implemented anonymous objects
+ * are possible in OSP. such an object isn't listed in the site
+ * so we should not remove it from the site.
+ */
+ fid = lu_object_fid(o);
+ if (fid_is_zero(fid)) {
+ LASSERT(top->loh_hash.next == NULL
+ && top->loh_hash.pprev == NULL);
+ LASSERT(list_empty(&top->loh_lru));
+ if (!atomic_dec_and_test(&top->loh_ref))
+ return;
+ list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
+ if (o->lo_ops->loo_object_release != NULL)
+ o->lo_ops->loo_object_release(env, o);
+ }
+ lu_object_free(env, orig);
+ return;
+ }
+
cfs_hash_bd_get(site->ls_obj_hash, &top->loh_fid, &bd);
bkt = cfs_hash_bd_extra_get(site->ls_obj_hash, &bd);
- if (!cfs_hash_bd_dec_and_lock(site->ls_obj_hash, &bd, &top->loh_ref)) {
- if (lu_object_is_dying(top)) {
+ if (!cfs_hash_bd_dec_and_lock(site->ls_obj_hash, &bd, &top->loh_ref)) {
+ if (lu_object_is_dying(top)) {
- /*
- * somebody may be waiting for this, currently only
- * used for cl_object, see cl_object_put_last().
- */
- cfs_waitq_broadcast(&bkt->lsb_marche_funebre);
- }
- return;
- }
+ /*
+ * somebody may be waiting for this, currently only
+ * used for cl_object, see cl_object_put_last().
+ */
+ wake_up_all(&bkt->lsb_marche_funebre);
+ }
+ return;
+ }
- LASSERT(bkt->lsb_busy > 0);
- bkt->lsb_busy--;
/*
* When last reference is released, iterate over object
* layers, and notify them that object is no longer busy.
*/
- cfs_list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
+ list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
if (o->lo_ops->loo_object_release != NULL)
o->lo_ops->loo_object_release(env, o);
}
- if (!lu_object_is_dying(top)) {
+ if (!lu_object_is_dying(top) &&
+ (lu_object_exists(orig) || lu_object_is_cl(orig))) {
+ LASSERT(list_empty(&top->loh_lru));
+ list_add_tail(&top->loh_lru, &bkt->lsb_lru);
+ bkt->lsb_lru_len++;
cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
return;
}
* and LRU lock, no race with concurrent object lookup is possible
* and we can safely destroy object below.
*/
- cfs_hash_bd_del_locked(site->ls_obj_hash, &bd, &top->loh_hash);
- cfs_list_del_init(&top->loh_lru);
+ if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags))
+ cfs_hash_bd_del_locked(site->ls_obj_hash, &bd, &top->loh_hash);
cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
/*
* Object was already removed from hash and lru above, can
EXPORT_SYMBOL(lu_object_put);
/**
+ * Put object and don't keep in cache. This is temporary solution for
+ * multi-site objects when its layering is not constant.
+ */
+void lu_object_put_nocache(const struct lu_env *env, struct lu_object *o)
+{
+ set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
+ return lu_object_put(env, o);
+}
+EXPORT_SYMBOL(lu_object_put_nocache);
+
+/**
+ * Kill the object and take it out of LRU cache.
+ * Currently used by client code for layout change.
+ */
+void lu_object_unhash(const struct lu_env *env, struct lu_object *o)
+{
+ struct lu_object_header *top;
+
+ top = o->lo_header;
+ set_bit(LU_OBJECT_HEARD_BANSHEE, &top->loh_flags);
+ if (!test_and_set_bit(LU_OBJECT_UNHASHED, &top->loh_flags)) {
+ cfs_hash_t *obj_hash = o->lo_dev->ld_site->ls_obj_hash;
+ cfs_hash_bd_t bd;
+
+ cfs_hash_bd_get_and_lock(obj_hash, &top->loh_fid, &bd, 1);
+ if (!list_empty(&top->loh_lru)) {
+ struct lu_site_bkt_data *bkt;
+
+ list_del_init(&top->loh_lru);
+ bkt = cfs_hash_bd_extra_get(obj_hash, &bd);
+ bkt->lsb_lru_len--;
+ }
+ cfs_hash_bd_del_locked(obj_hash, &bd, &top->loh_hash);
+ cfs_hash_bd_unlock(obj_hash, &bd, 1);
+ }
+}
+EXPORT_SYMBOL(lu_object_unhash);
+
+/**
* Allocate new object.
*
* This follows object creation protocol, described in the comment within
* struct lu_device_operations definition.
*/
static struct lu_object *lu_object_alloc(const struct lu_env *env,
- struct lu_device *dev,
- const struct lu_fid *f,
- const struct lu_object_conf *conf)
-{
- struct lu_object *scan;
- struct lu_object *top;
- cfs_list_t *layers;
- int clean;
- int result;
- ENTRY;
-
- /*
- * Create top-level object slice. This will also create
- * lu_object_header.
- */
- top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
- if (top == NULL)
- RETURN(ERR_PTR(-ENOMEM));
+ struct lu_device *dev,
+ const struct lu_fid *f,
+ const struct lu_object_conf *conf)
+{
+ struct lu_object *scan;
+ struct lu_object *top;
+ struct list_head *layers;
+ unsigned int init_mask = 0;
+ unsigned int init_flag;
+ int clean;
+ int result;
+ ENTRY;
+
+ /*
+ * Create top-level object slice. This will also create
+ * lu_object_header.
+ */
+ top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
+ if (top == NULL)
+ RETURN(ERR_PTR(-ENOMEM));
+ if (IS_ERR(top))
+ RETURN(top);
/*
* This is the only place where object fid is assigned. It's constant
* after this point.
*/
- LASSERT(fid_is_igif(f) || fid_ver(f) == 0);
top->lo_header->loh_fid = *f;
layers = &top->lo_header->loh_layers;
- do {
- /*
- * Call ->loo_object_init() repeatedly, until no more new
- * object slices are created.
- */
- clean = 1;
- cfs_list_for_each_entry(scan, layers, lo_linkage) {
- if (scan->lo_flags & LU_OBJECT_ALLOCATED)
- continue;
- clean = 0;
- scan->lo_header = top->lo_header;
- result = scan->lo_ops->loo_object_init(env, scan, conf);
- if (result != 0) {
- lu_object_free(env, top);
- RETURN(ERR_PTR(result));
- }
- scan->lo_flags |= LU_OBJECT_ALLOCATED;
- }
- } while (!clean);
- cfs_list_for_each_entry_reverse(scan, layers, lo_linkage) {
+ do {
+ /*
+ * Call ->loo_object_init() repeatedly, until no more new
+ * object slices are created.
+ */
+ clean = 1;
+ init_flag = 1;
+ list_for_each_entry(scan, layers, lo_linkage) {
+ if (init_mask & init_flag)
+ goto next;
+ clean = 0;
+ scan->lo_header = top->lo_header;
+ result = scan->lo_ops->loo_object_init(env, scan, conf);
+ if (result != 0) {
+ lu_object_free(env, top);
+ RETURN(ERR_PTR(result));
+ }
+ init_mask |= init_flag;
+next:
+ init_flag <<= 1;
+ }
+ } while (!clean);
+
+ list_for_each_entry_reverse(scan, layers, lo_linkage) {
if (scan->lo_ops->loo_object_start != NULL) {
result = scan->lo_ops->loo_object_start(env, scan);
if (result != 0) {
*/
static void lu_object_free(const struct lu_env *env, struct lu_object *o)
{
- struct lu_site_bkt_data *bkt;
- struct lu_site *site;
- struct lu_object *scan;
- cfs_list_t *layers;
- cfs_list_t splice;
+ struct lu_site_bkt_data *bkt;
+ struct lu_site *site;
+ struct lu_object *scan;
+ struct list_head *layers;
+ struct list_head splice;
site = o->lo_dev->ld_site;
layers = &o->lo_header->loh_layers;
/*
* First call ->loo_object_delete() method to release all resources.
*/
- cfs_list_for_each_entry_reverse(scan, layers, lo_linkage) {
+ list_for_each_entry_reverse(scan, layers, lo_linkage) {
if (scan->lo_ops->loo_object_delete != NULL)
scan->lo_ops->loo_object_delete(env, scan);
}
* necessary, because lu_object_header is freed together with the
* top-level slice.
*/
- CFS_INIT_LIST_HEAD(&splice);
- cfs_list_splice_init(layers, &splice);
- while (!cfs_list_empty(&splice)) {
- /*
- * Free layers in bottom-to-top order, so that object header
- * lives as long as possible and ->loo_object_free() methods
- * can look at its contents.
- */
- o = container_of0(splice.prev, struct lu_object, lo_linkage);
- cfs_list_del_init(&o->lo_linkage);
- LASSERT(o->lo_ops->loo_object_free != NULL);
- o->lo_ops->loo_object_free(env, o);
- }
-
- if (cfs_waitq_active(&bkt->lsb_marche_funebre))
- cfs_waitq_broadcast(&bkt->lsb_marche_funebre);
+ INIT_LIST_HEAD(&splice);
+ list_splice_init(layers, &splice);
+ while (!list_empty(&splice)) {
+ /*
+ * Free layers in bottom-to-top order, so that object header
+ * lives as long as possible and ->loo_object_free() methods
+ * can look at its contents.
+ */
+ o = container_of0(splice.prev, struct lu_object, lo_linkage);
+ list_del_init(&o->lo_linkage);
+ LASSERT(o->lo_ops->loo_object_free != NULL);
+ o->lo_ops->loo_object_free(env, o);
+ }
+
+ if (waitqueue_active(&bkt->lsb_marche_funebre))
+ wake_up_all(&bkt->lsb_marche_funebre);
}
/**
struct lu_site_bkt_data *bkt;
cfs_hash_bd_t bd;
cfs_hash_bd_t bd2;
- cfs_list_t dispose;
- int did_sth;
- int start;
+ struct list_head dispose;
+ int did_sth;
+ unsigned int start;
int count;
int bnr;
- int i;
+ unsigned int i;
+
+ if (OBD_FAIL_CHECK(OBD_FAIL_OBD_NO_LRU))
+ RETURN(0);
- CFS_INIT_LIST_HEAD(&dispose);
+ INIT_LIST_HEAD(&dispose);
/*
* Under LRU list lock, scan LRU list and move unreferenced objects to
* the dispose list, removing them from LRU and hash table.
*/
start = s->ls_purge_start;
- bnr = (nr == ~0) ? -1 : nr / CFS_HASH_NBKT(s->ls_obj_hash) + 1;
+ bnr = (nr == ~0) ? -1 : nr / (int)CFS_HASH_NBKT(s->ls_obj_hash) + 1;
again:
+ /*
+ * It doesn't make any sense to make purge threads parallel, that can
+ * only bring troubles to us. See LU-5331.
+ */
+ mutex_lock(&s->ls_purge_mutex);
did_sth = 0;
cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
if (i < start)
cfs_hash_bd_lock(s->ls_obj_hash, &bd, 1);
bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
- cfs_list_for_each_entry_safe(h, temp, &bkt->lsb_lru, loh_lru) {
- /*
- * Objects are sorted in lru order, and "busy"
- * objects (ones with h->loh_ref > 0) naturally tend to
- * live near hot end that we scan last. Unfortunately,
- * sites usually have small (less then ten) number of
- * busy yet rarely accessed objects (some global
- * objects, accessed directly through pointers,
- * bypassing hash table).
- * Currently algorithm scans them over and over again.
- * Probably we should move busy objects out of LRU,
- * or we can live with that.
- */
- if (cfs_atomic_read(&h->loh_ref) > 0)
- continue;
+ list_for_each_entry_safe(h, temp, &bkt->lsb_lru, loh_lru) {
+ LASSERT(atomic_read(&h->loh_ref) == 0);
cfs_hash_bd_get(s->ls_obj_hash, &h->loh_fid, &bd2);
LASSERT(bd.bd_bucket == bd2.bd_bucket);
cfs_hash_bd_del_locked(s->ls_obj_hash,
&bd2, &h->loh_hash);
- cfs_list_move(&h->loh_lru, &dispose);
+ list_move(&h->loh_lru, &dispose);
+ bkt->lsb_lru_len--;
if (did_sth == 0)
did_sth = 1;
if (count > 0 && --count == 0)
break;
- }
- cfs_hash_bd_unlock(s->ls_obj_hash, &bd, 1);
- cfs_cond_resched();
- /*
- * Free everything on the dispose list. This is safe against
- * races due to the reasons described in lu_object_put().
- */
- while (!cfs_list_empty(&dispose)) {
- h = container_of0(dispose.next,
- struct lu_object_header, loh_lru);
- cfs_list_del_init(&h->loh_lru);
- lu_object_free(env, lu_object_top(h));
- lprocfs_counter_incr(s->ls_stats, LU_SS_LRU_PURGED);
- }
+ }
+ cfs_hash_bd_unlock(s->ls_obj_hash, &bd, 1);
+ cond_resched();
+ /*
+ * Free everything on the dispose list. This is safe against
+ * races due to the reasons described in lu_object_put().
+ */
+ while (!list_empty(&dispose)) {
+ h = container_of0(dispose.next,
+ struct lu_object_header, loh_lru);
+ list_del_init(&h->loh_lru);
+ lu_object_free(env, lu_object_top(h));
+ lprocfs_counter_incr(s->ls_stats, LU_SS_LRU_PURGED);
+ }
if (nr == 0)
break;
}
+ mutex_unlock(&s->ls_purge_mutex);
if (nr != 0 && did_sth && start != 0) {
start = 0; /* restart from the first bucket */
*
* XXX overflow is not handled correctly.
*/
- LU_CDEBUG_LINE = 256
+ LU_CDEBUG_LINE = 512
};
struct lu_cdebug_data {
* Key, holding temporary buffer. This key is registered very early by
* lu_global_init().
*/
-struct lu_context_key lu_global_key = {
- .lct_tags = LCT_MD_THREAD|LCT_DT_THREAD|LCT_CL_THREAD,
- .lct_init = lu_global_key_init,
- .lct_fini = lu_global_key_fini
+static struct lu_context_key lu_global_key = {
+ .lct_tags = LCT_MD_THREAD | LCT_DT_THREAD |
+ LCT_MG_THREAD | LCT_CL_THREAD | LCT_LOCAL,
+ .lct_init = lu_global_key_init,
+ .lct_fini = lu_global_key_fini
};
/**
int lu_cdebug_printer(const struct lu_env *env,
void *cookie, const char *format, ...)
{
- struct lu_cdebug_print_info *info = cookie;
- struct lu_cdebug_data *key;
+ struct libcfs_debug_msg_data *msgdata = cookie;
+ struct lu_cdebug_data *key;
int used;
int complete;
va_list args;
vsnprintf(key->lck_area + used,
ARRAY_SIZE(key->lck_area) - used, format, args);
if (complete) {
- if (cfs_cdebug_show(info->lpi_mask, info->lpi_subsys))
- libcfs_debug_msg(NULL, info->lpi_subsys, info->lpi_mask,
- (char *)info->lpi_file, info->lpi_fn,
- info->lpi_line, "%s", key->lck_area);
+ if (cfs_cdebug_show(msgdata->msg_mask, msgdata->msg_subsys))
+ libcfs_debug_msg(msgdata, "%s\n", key->lck_area);
key->lck_area[0] = 0;
}
va_end(args);
lu_printer_t printer,
const struct lu_object_header *hdr)
{
- (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
- hdr, hdr->loh_flags, cfs_atomic_read(&hdr->loh_ref),
- PFID(&hdr->loh_fid),
- cfs_hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
- cfs_list_empty((cfs_list_t *)&hdr->loh_lru) ? \
- "" : " lru",
- hdr->loh_attr & LOHA_EXISTS ? " exist":"");
+ (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
+ hdr, hdr->loh_flags, atomic_read(&hdr->loh_ref),
+ PFID(&hdr->loh_fid),
+ hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
+ list_empty((struct list_head *)&hdr->loh_lru) ? \
+ "" : " lru",
+ hdr->loh_attr & LOHA_EXISTS ? " exist" : "");
}
EXPORT_SYMBOL(lu_object_header_print);
* Print human readable representation of the \a o to the \a printer.
*/
void lu_object_print(const struct lu_env *env, void *cookie,
- lu_printer_t printer, const struct lu_object *o)
+ lu_printer_t printer, const struct lu_object *o)
{
- static const char ruler[] = "........................................";
- struct lu_object_header *top;
- int depth;
+ static const char ruler[] = "........................................";
+ struct lu_object_header *top;
+ int depth = 4;
- top = o->lo_header;
- lu_object_header_print(env, cookie, printer, top);
- (*printer)(env, cookie, "{ \n");
- cfs_list_for_each_entry(o, &top->loh_layers, lo_linkage) {
- depth = o->lo_depth + 4;
+ top = o->lo_header;
+ lu_object_header_print(env, cookie, printer, top);
+ (*printer)(env, cookie, "{\n");
- /*
- * print `.' \a depth times followed by type name and address
- */
- (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
- o->lo_dev->ld_type->ldt_name, o);
- if (o->lo_ops->loo_object_print != NULL)
- o->lo_ops->loo_object_print(env, cookie, printer, o);
- (*printer)(env, cookie, "\n");
- }
- (*printer)(env, cookie, "} header@%p\n", top);
+ list_for_each_entry(o, &top->loh_layers, lo_linkage) {
+ /*
+ * print `.' \a depth times followed by type name and address
+ */
+ (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
+ o->lo_dev->ld_type->ldt_name, o);
+
+ if (o->lo_ops->loo_object_print != NULL)
+ (*o->lo_ops->loo_object_print)(env, cookie, printer, o);
+
+ (*printer)(env, cookie, "\n");
+ }
+
+ (*printer)(env, cookie, "} header@%p\n", top);
}
EXPORT_SYMBOL(lu_object_print);
struct lu_object_header *top;
top = o->lo_header;
- cfs_list_for_each_entry(o, &top->loh_layers, lo_linkage) {
+ list_for_each_entry(o, &top->loh_layers, lo_linkage) {
if (o->lo_ops->loo_object_invariant != NULL &&
!o->lo_ops->loo_object_invariant(o))
return 0;
}
return 1;
}
-EXPORT_SYMBOL(lu_object_invariant);
static struct lu_object *htable_lookup(struct lu_site *s,
- cfs_hash_bd_t *bd,
- const struct lu_fid *f,
- cfs_waitlink_t *waiter,
- __u64 *version)
+ cfs_hash_bd_t *bd,
+ const struct lu_fid *f,
+ wait_queue_t *waiter,
+ __u64 *version)
{
- struct lu_site_bkt_data *bkt;
- struct lu_object_header *h;
- cfs_hlist_node_t *hnode;
- __u64 ver = cfs_hash_bd_version_get(bd);
+ struct lu_site_bkt_data *bkt;
+ struct lu_object_header *h;
+ struct hlist_node *hnode;
+ __u64 ver = cfs_hash_bd_version_get(bd);
if (*version == ver)
- return NULL;
+ return ERR_PTR(-ENOENT);
*version = ver;
bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, bd);
- /* cfs_hash_bd_lookup_intent is a somehow "internal" function
- * of cfs_hash, but we don't want refcount on object right now */
- hnode = cfs_hash_bd_lookup_locked(s->ls_obj_hash, bd, (void *)f);
+ /* cfs_hash_bd_peek_locked is a somehow "internal" function
+ * of cfs_hash, it doesn't add refcount on object. */
+ hnode = cfs_hash_bd_peek_locked(s->ls_obj_hash, bd, (void *)f);
if (hnode == NULL) {
lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
- return NULL;
+ return ERR_PTR(-ENOENT);
}
h = container_of0(hnode, struct lu_object_header, loh_hash);
if (likely(!lu_object_is_dying(h))) {
+ cfs_hash_get(s->ls_obj_hash, hnode);
lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_HIT);
+ if (!list_empty(&h->loh_lru)) {
+ list_del_init(&h->loh_lru);
+ bkt->lsb_lru_len--;
+ }
return lu_object_top(h);
}
* returned (to assure that references to dying objects are eventually
* drained), and moreover, lookup has to wait until object is freed.
*/
- cfs_atomic_dec(&h->loh_ref);
- cfs_waitlink_init(waiter);
- cfs_waitq_add(&bkt->lsb_marche_funebre, waiter);
- cfs_set_current_state(CFS_TASK_UNINT);
- lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_DEATH_RACE);
- return ERR_PTR(-EAGAIN);
+ if (likely(waiter != NULL)) {
+ init_waitqueue_entry_current(waiter);
+ add_wait_queue(&bkt->lsb_marche_funebre, waiter);
+ set_current_state(TASK_UNINTERRUPTIBLE);
+ lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_DEATH_RACE);
+ }
+
+ return ERR_PTR(-EAGAIN);
}
/**
}
EXPORT_SYMBOL(lu_object_find);
+/*
+ * Limit the lu_object cache to a maximum of lu_cache_nr objects. Because
+ * the calculation for the number of objects to reclaim is not covered by
+ * a lock the maximum number of objects is capped by LU_CACHE_MAX_ADJUST.
+ * This ensures that many concurrent threads will not accidentally purge
+ * the entire cache.
+ */
+static void lu_object_limit(const struct lu_env *env,
+ struct lu_device *dev)
+{
+ __u64 size, nr;
+
+ if (lu_cache_nr == LU_CACHE_NR_UNLIMITED)
+ return;
+
+ size = cfs_hash_size_get(dev->ld_site->ls_obj_hash);
+ nr = (__u64)lu_cache_nr;
+ if (size > nr)
+ lu_site_purge(env, dev->ld_site,
+ MIN(size - nr, LU_CACHE_NR_MAX_ADJUST));
+
+ return;
+}
+
+static struct lu_object *lu_object_new(const struct lu_env *env,
+ struct lu_device *dev,
+ const struct lu_fid *f,
+ const struct lu_object_conf *conf)
+{
+ struct lu_object *o;
+ cfs_hash_t *hs;
+ cfs_hash_bd_t bd;
+
+ o = lu_object_alloc(env, dev, f, conf);
+ if (unlikely(IS_ERR(o)))
+ return o;
+
+ hs = dev->ld_site->ls_obj_hash;
+ cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
+ cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
+ cfs_hash_bd_unlock(hs, &bd, 1);
+
+ lu_object_limit(env, dev);
+
+ return o;
+}
+
/**
* Core logic of lu_object_find*() functions.
*/
static struct lu_object *lu_object_find_try(const struct lu_env *env,
- struct lu_device *dev,
- const struct lu_fid *f,
- const struct lu_object_conf *conf,
- cfs_waitlink_t *waiter)
-{
- struct lu_object *o;
- struct lu_object *shadow;
- struct lu_site *s;
- cfs_hash_t *hs;
- cfs_hash_bd_t bd;
- __u64 version = 0;
+ struct lu_device *dev,
+ const struct lu_fid *f,
+ const struct lu_object_conf *conf,
+ wait_queue_t *waiter)
+{
+ struct lu_object *o;
+ struct lu_object *shadow;
+ struct lu_site *s;
+ cfs_hash_t *hs;
+ cfs_hash_bd_t bd;
+ __u64 version = 0;
/*
* This uses standard index maintenance protocol:
* - unlock index;
* - return object.
*
+ * For "LOC_F_NEW" case, we are sure the object is new established.
+ * It is unnecessary to perform lookup-alloc-lookup-insert, instead,
+ * just alloc and insert directly.
+ *
* If dying object is found during index search, add @waiter to the
* site wait-queue and return ERR_PTR(-EAGAIN).
*/
+ if (conf != NULL && conf->loc_flags & LOC_F_NEW)
+ return lu_object_new(env, dev, f, conf);
+
s = dev->ld_site;
hs = s->ls_obj_hash;
cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
o = htable_lookup(s, &bd, f, waiter, &version);
cfs_hash_bd_unlock(hs, &bd, 1);
- if (o != NULL)
+ if (!IS_ERR(o) || PTR_ERR(o) != -ENOENT)
return o;
/*
cfs_hash_bd_lock(hs, &bd, 1);
shadow = htable_lookup(s, &bd, f, waiter, &version);
- if (likely(shadow == NULL)) {
- struct lu_site_bkt_data *bkt;
-
- bkt = cfs_hash_bd_extra_get(hs, &bd);
+ if (likely(IS_ERR(shadow) && PTR_ERR(shadow) == -ENOENT)) {
cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
- cfs_list_add_tail(&o->lo_header->loh_lru, &bkt->lsb_lru);
- bkt->lsb_busy++;
cfs_hash_bd_unlock(hs, &bd, 1);
+
+ lu_object_limit(env, dev);
+
return o;
}
* objects of different "stacking" to be created within the same site.
*/
struct lu_object *lu_object_find_at(const struct lu_env *env,
- struct lu_device *dev,
- const struct lu_fid *f,
- const struct lu_object_conf *conf)
-{
- struct lu_site_bkt_data *bkt;
- struct lu_object *obj;
- cfs_waitlink_t wait;
-
- while (1) {
- obj = lu_object_find_try(env, dev, f, conf, &wait);
- if (obj != ERR_PTR(-EAGAIN))
- return obj;
- /*
- * lu_object_find_try() already added waiter into the
- * wait queue.
- */
- cfs_waitq_wait(&wait, CFS_TASK_UNINT);
- bkt = lu_site_bkt_from_fid(dev->ld_site, (void *)f);
- cfs_waitq_del(&bkt->lsb_marche_funebre, &wait);
- }
+ struct lu_device *dev,
+ const struct lu_fid *f,
+ const struct lu_object_conf *conf)
+{
+ struct lu_site_bkt_data *bkt;
+ struct lu_object *obj;
+ wait_queue_t wait;
+
+ if (conf != NULL && conf->loc_flags & LOC_F_NOWAIT)
+ return lu_object_find_try(env, dev, f, conf, NULL);
+
+ while (1) {
+ obj = lu_object_find_try(env, dev, f, conf, &wait);
+ if (obj != ERR_PTR(-EAGAIN))
+ return obj;
+ /*
+ * lu_object_find_try() already added waiter into the
+ * wait queue.
+ */
+ waitq_wait(&wait, TASK_UNINTERRUPTIBLE);
+ bkt = lu_site_bkt_from_fid(dev->ld_site, (void *)f);
+ remove_wait_queue(&bkt->lsb_marche_funebre, &wait);
+ }
}
EXPORT_SYMBOL(lu_object_find_at);
const struct lu_fid *f,
const struct lu_object_conf *conf)
{
- struct lu_object *top;
- struct lu_object *obj;
+ struct lu_object *top;
+ struct lu_object *obj;
+
+ top = lu_object_find(env, dev, f, conf);
+ if (IS_ERR(top))
+ return top;
- top = lu_object_find(env, dev, f, conf);
- if (!IS_ERR(top)) {
- obj = lu_object_locate(top->lo_header, dev->ld_type);
- if (obj == NULL)
- lu_object_put(env, top);
- } else
- obj = top;
- return obj;
+ obj = lu_object_locate(top->lo_header, dev->ld_type);
+ if (unlikely(obj == NULL)) {
+ lu_object_put(env, top);
+ obj = ERR_PTR(-ENOENT);
+ }
+
+ return obj;
}
EXPORT_SYMBOL(lu_object_find_slice);
/**
* Global list of all device types.
*/
-static CFS_LIST_HEAD(lu_device_types);
+static struct list_head lu_device_types;
int lu_device_type_init(struct lu_device_type *ldt)
{
- int result;
+ int result = 0;
- CFS_INIT_LIST_HEAD(&ldt->ldt_linkage);
- result = ldt->ldt_ops->ldto_init(ldt);
- if (result == 0)
- cfs_list_add(&ldt->ldt_linkage, &lu_device_types);
- return result;
+ atomic_set(&ldt->ldt_device_nr, 0);
+ INIT_LIST_HEAD(&ldt->ldt_linkage);
+ if (ldt->ldt_ops->ldto_init)
+ result = ldt->ldt_ops->ldto_init(ldt);
+
+ if (result == 0) {
+ spin_lock(&obd_types_lock);
+ list_add(&ldt->ldt_linkage, &lu_device_types);
+ spin_unlock(&obd_types_lock);
+ }
+
+ return result;
}
EXPORT_SYMBOL(lu_device_type_init);
void lu_device_type_fini(struct lu_device_type *ldt)
{
- cfs_list_del_init(&ldt->ldt_linkage);
- ldt->ldt_ops->ldto_fini(ldt);
+ spin_lock(&obd_types_lock);
+ list_del_init(&ldt->ldt_linkage);
+ spin_unlock(&obd_types_lock);
+ if (ldt->ldt_ops->ldto_fini)
+ ldt->ldt_ops->ldto_fini(ldt);
}
EXPORT_SYMBOL(lu_device_type_fini);
-void lu_types_stop(void)
-{
- struct lu_device_type *ldt;
-
- cfs_list_for_each_entry(ldt, &lu_device_types, ldt_linkage) {
- if (ldt->ldt_device_nr == 0)
- ldt->ldt_ops->ldto_stop(ldt);
- }
-}
-EXPORT_SYMBOL(lu_types_stop);
-
/**
* Global list of all sites on this node
*/
-static CFS_LIST_HEAD(lu_sites);
-static CFS_DECLARE_MUTEX(lu_sites_guard);
+static struct list_head lu_sites;
+static DEFINE_MUTEX(lu_sites_guard);
/**
* Global environment used by site shrinker.
static int
lu_site_obj_print(cfs_hash_t *hs, cfs_hash_bd_t *bd,
- cfs_hlist_node_t *hnode, void *data)
+ struct hlist_node *hnode, void *data)
{
- struct lu_site_print_arg *arg = (struct lu_site_print_arg *)data;
- struct lu_object_header *h;
+ struct lu_site_print_arg *arg = (struct lu_site_print_arg *)data;
+ struct lu_object_header *h;
- h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
- if (!cfs_list_empty(&h->loh_layers)) {
- const struct lu_object *o;
+ h = hlist_entry(hnode, struct lu_object_header, loh_hash);
+ if (!list_empty(&h->loh_layers)) {
+ const struct lu_object *o;
- o = lu_object_top(h);
- lu_object_print(arg->lsp_env, arg->lsp_cookie,
- arg->lsp_printer, o);
- } else {
- lu_object_header_print(arg->lsp_env, arg->lsp_cookie,
- arg->lsp_printer, h);
- }
- return 0;
+ o = lu_object_top(h);
+ lu_object_print(arg->lsp_env, arg->lsp_cookie,
+ arg->lsp_printer, o);
+ } else {
+ lu_object_header_print(arg->lsp_env, arg->lsp_cookie,
+ arg->lsp_printer, h);
+ }
+ return 0;
}
/**
}
EXPORT_SYMBOL(lu_site_print);
-enum {
- LU_CACHE_PERCENT_MAX = 50,
- LU_CACHE_PERCENT_DEFAULT = 20
-};
-
-static unsigned int lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
-CFS_MODULE_PARM(lu_cache_percent, "i", int, 0644,
- "Percentage of memory to be used as lu_object cache");
-
/**
* Return desired hash table order.
*/
-static int lu_htable_order(void)
-{
- unsigned long cache_size;
- int bits;
+static unsigned long lu_htable_order(struct lu_device *top)
+{
+ unsigned long cache_size;
+ unsigned long bits;
+
+ /*
+ * For ZFS based OSDs the cache should be disabled by default. This
+ * allows the ZFS ARC maximum flexibility in determining what buffers
+ * to cache. If Lustre has objects or buffer which it wants to ensure
+ * always stay cached it must maintain a hold on them.
+ */
+ if (strcmp(top->ld_type->ldt_name, LUSTRE_OSD_ZFS_NAME) == 0) {
+ lu_cache_percent = 1;
+ lu_cache_nr = LU_CACHE_NR_ZFS_LIMIT;
+ return LU_SITE_BITS_MIN;
+ }
/*
* Calculate hash table size, assuming that we want reasonable
*
* Size of lu_object is (arbitrary) taken as 1K (together with inode).
*/
- cache_size = cfs_num_physpages;
+ cache_size = totalram_pages;
#if BITS_PER_LONG == 32
/* limit hashtable size for lowmem systems to low RAM */
- if (cache_size > 1 << (30 - CFS_PAGE_SHIFT))
- cache_size = 1 << (30 - CFS_PAGE_SHIFT) * 3 / 4;
+ if (cache_size > 1 << (30 - PAGE_CACHE_SHIFT))
+ cache_size = 1 << (30 - PAGE_CACHE_SHIFT) * 3 / 4;
#endif
/* clear off unreasonable cache setting. */
lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
}
cache_size = cache_size / 100 * lu_cache_percent *
- (CFS_PAGE_SIZE / 1024);
+ (PAGE_CACHE_SIZE / 1024);
for (bits = 1; (1 << bits) < cache_size; ++bits) {
;
}
static unsigned lu_obj_hop_hash(cfs_hash_t *hs,
- const void *key, unsigned mask)
+ const void *key, unsigned mask)
{
- struct lu_fid *fid = (struct lu_fid *)key;
- unsigned hash;
+ struct lu_fid *fid = (struct lu_fid *)key;
+ __u32 hash;
- hash = (fid_seq(fid) + fid_oid(fid)) & (CFS_HASH_NBKT(hs) - 1);
- hash += fid_hash(fid, hs->hs_bkt_bits) << hs->hs_bkt_bits;
- return hash & mask;
+ hash = fid_flatten32(fid);
+ hash += (hash >> 4) + (hash << 12); /* mixing oid and seq */
+ hash = hash_long(hash, hs->hs_bkt_bits);
+
+ /* give me another random factor */
+ hash -= hash_long((unsigned long)hs, fid_oid(fid) % 11 + 3);
+
+ hash <<= hs->hs_cur_bits - hs->hs_bkt_bits;
+ hash |= (fid_seq(fid) + fid_oid(fid)) & (CFS_HASH_NBKT(hs) - 1);
+
+ return hash & mask;
}
-static void *lu_obj_hop_object(cfs_hlist_node_t *hnode)
+static void *lu_obj_hop_object(struct hlist_node *hnode)
{
- return cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
+ return hlist_entry(hnode, struct lu_object_header, loh_hash);
}
-static void *lu_obj_hop_key(cfs_hlist_node_t *hnode)
+static void *lu_obj_hop_key(struct hlist_node *hnode)
{
- struct lu_object_header *h;
+ struct lu_object_header *h;
- h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
- return &h->loh_fid;
+ h = hlist_entry(hnode, struct lu_object_header, loh_hash);
+ return &h->loh_fid;
}
-static int lu_obj_hop_keycmp(const void *key, cfs_hlist_node_t *hnode)
+static int lu_obj_hop_keycmp(const void *key, struct hlist_node *hnode)
{
- struct lu_object_header *h;
+ struct lu_object_header *h;
- h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
- return lu_fid_eq(&h->loh_fid, (struct lu_fid *)key);
+ h = hlist_entry(hnode, struct lu_object_header, loh_hash);
+ return lu_fid_eq(&h->loh_fid, (struct lu_fid *)key);
}
-static void lu_obj_hop_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+static void lu_obj_hop_get(cfs_hash_t *hs, struct hlist_node *hnode)
{
- struct lu_object_header *h;
-
- h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
- if (cfs_atomic_add_return(1, &h->loh_ref) == 1) {
- struct lu_site_bkt_data *bkt;
- cfs_hash_bd_t bd;
+ struct lu_object_header *h;
- cfs_hash_bd_get(hs, &h->loh_fid, &bd);
- bkt = cfs_hash_bd_extra_get(hs, &bd);
- bkt->lsb_busy++;
- }
+ h = hlist_entry(hnode, struct lu_object_header, loh_hash);
+ atomic_inc(&h->loh_ref);
}
-static void lu_obj_hop_put_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
+static void lu_obj_hop_put_locked(cfs_hash_t *hs, struct hlist_node *hnode)
{
LBUG(); /* we should never called it */
}
-cfs_hash_ops_t lu_site_hash_ops = {
+static cfs_hash_ops_t lu_site_hash_ops = {
.hs_hash = lu_obj_hop_hash,
.hs_key = lu_obj_hop_key,
.hs_keycmp = lu_obj_hop_keycmp,
.hs_put_locked = lu_obj_hop_put_locked,
};
-/**
- * Initialize site \a s, with \a d as the top level device.
- */
-#define LU_SITE_BITS_MIN 12
-#define LU_SITE_BITS_MAX 23
-/**
- * total 128 buckets, we don't want too many buckets because:
- * - consume too much memory
- * - avoid unbalanced LRU list
- */
-#define LU_SITE_BKT_BITS 7
-
-int lu_site_init(struct lu_site *s, struct lu_device *top)
+void lu_dev_add_linkage(struct lu_site *s, struct lu_device *d)
{
- struct lu_site_bkt_data *bkt;
- cfs_hash_bd_t bd;
- int bits;
- int i;
- ENTRY;
-
- memset(s, 0, sizeof *s);
- bits = lu_htable_order();
- for (bits = min(max(LU_SITE_BITS_MIN, bits), LU_SITE_BITS_MAX);
- bits >= LU_SITE_BITS_MIN; bits--) {
- s->ls_obj_hash = cfs_hash_create("lu_site", bits, bits,
- bits - LU_SITE_BKT_BITS,
- sizeof(*bkt), 0, 0,
- &lu_site_hash_ops,
- CFS_HASH_SPIN_BKTLOCK |
- CFS_HASH_NO_ITEMREF |
- CFS_HASH_DEPTH |
- CFS_HASH_ASSERT_EMPTY);
- if (s->ls_obj_hash != NULL)
- break;
- }
+ spin_lock(&s->ls_ld_lock);
+ if (list_empty(&d->ld_linkage))
+ list_add(&d->ld_linkage, &s->ls_ld_linkage);
+ spin_unlock(&s->ls_ld_lock);
+}
+EXPORT_SYMBOL(lu_dev_add_linkage);
- if (s->ls_obj_hash == NULL) {
- CERROR("failed to create lu_site hash with bits: %d\n", bits);
- return -ENOMEM;
- }
+void lu_dev_del_linkage(struct lu_site *s, struct lu_device *d)
+{
+ spin_lock(&s->ls_ld_lock);
+ list_del_init(&d->ld_linkage);
+ spin_unlock(&s->ls_ld_lock);
+}
+EXPORT_SYMBOL(lu_dev_del_linkage);
- cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
- bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
- CFS_INIT_LIST_HEAD(&bkt->lsb_lru);
- cfs_waitq_init(&bkt->lsb_marche_funebre);
- }
+/**
+ * Initialize site \a s, with \a d as the top level device.
+ */
+int lu_site_init(struct lu_site *s, struct lu_device *top)
+{
+ struct lu_site_bkt_data *bkt;
+ cfs_hash_bd_t bd;
+ char name[16];
+ unsigned long bits;
+ unsigned int i;
+ ENTRY;
+
+ memset(s, 0, sizeof *s);
+ mutex_init(&s->ls_purge_mutex);
+ bits = lu_htable_order(top);
+ snprintf(name, sizeof(name), "lu_site_%s", top->ld_type->ldt_name);
+ for (bits = clamp_t(typeof(bits), bits,
+ LU_SITE_BITS_MIN, LU_SITE_BITS_MAX);
+ bits >= LU_SITE_BITS_MIN; bits--) {
+ s->ls_obj_hash = cfs_hash_create(name, bits, bits,
+ bits - LU_SITE_BKT_BITS,
+ sizeof(*bkt), 0, 0,
+ &lu_site_hash_ops,
+ CFS_HASH_SPIN_BKTLOCK |
+ CFS_HASH_NO_ITEMREF |
+ CFS_HASH_DEPTH |
+ CFS_HASH_ASSERT_EMPTY |
+ CFS_HASH_COUNTER);
+ if (s->ls_obj_hash != NULL)
+ break;
+ }
+
+ if (s->ls_obj_hash == NULL) {
+ CERROR("failed to create lu_site hash with bits: %lu\n", bits);
+ return -ENOMEM;
+ }
+
+ cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
+ bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
+ INIT_LIST_HEAD(&bkt->lsb_lru);
+ init_waitqueue_head(&bkt->lsb_marche_funebre);
+ }
s->ls_stats = lprocfs_alloc_stats(LU_SS_LAST_STAT, 0);
if (s->ls_stats == NULL) {
lprocfs_counter_init(s->ls_stats, LU_SS_LRU_PURGED,
0, "lru_purged", "lru_purged");
- CFS_INIT_LIST_HEAD(&s->ls_linkage);
+ INIT_LIST_HEAD(&s->ls_linkage);
s->ls_top_dev = top;
top->ld_site = s;
lu_device_get(top);
lu_ref_add(&top->ld_reference, "site-top", s);
- RETURN(0);
+ INIT_LIST_HEAD(&s->ls_ld_linkage);
+ spin_lock_init(&s->ls_ld_lock);
+
+ lu_dev_add_linkage(s, top);
+
+ RETURN(0);
}
EXPORT_SYMBOL(lu_site_init);
*/
void lu_site_fini(struct lu_site *s)
{
- cfs_down(&lu_sites_guard);
- cfs_list_del_init(&s->ls_linkage);
- cfs_up(&lu_sites_guard);
+ mutex_lock(&lu_sites_guard);
+ list_del_init(&s->ls_linkage);
+ mutex_unlock(&lu_sites_guard);
if (s->ls_obj_hash != NULL) {
cfs_hash_putref(s->ls_obj_hash);
int lu_site_init_finish(struct lu_site *s)
{
int result;
- cfs_down(&lu_sites_guard);
+ mutex_lock(&lu_sites_guard);
result = lu_context_refill(&lu_shrink_env.le_ctx);
if (result == 0)
- cfs_list_add(&s->ls_linkage, &lu_sites);
- cfs_up(&lu_sites_guard);
+ list_add(&s->ls_linkage, &lu_sites);
+ mutex_unlock(&lu_sites_guard);
return result;
}
EXPORT_SYMBOL(lu_site_init_finish);
*/
void lu_device_get(struct lu_device *d)
{
- cfs_atomic_inc(&d->ld_ref);
+ atomic_inc(&d->ld_ref);
}
EXPORT_SYMBOL(lu_device_get);
*/
void lu_device_put(struct lu_device *d)
{
- LASSERT(cfs_atomic_read(&d->ld_ref) > 0);
- cfs_atomic_dec(&d->ld_ref);
+ LASSERT(atomic_read(&d->ld_ref) > 0);
+ atomic_dec(&d->ld_ref);
}
EXPORT_SYMBOL(lu_device_put);
*/
int lu_device_init(struct lu_device *d, struct lu_device_type *t)
{
- if (t->ldt_device_nr++ == 0 && t->ldt_ops->ldto_start != NULL)
- t->ldt_ops->ldto_start(t);
- memset(d, 0, sizeof *d);
- cfs_atomic_set(&d->ld_ref, 0);
- d->ld_type = t;
- lu_ref_init(&d->ld_reference);
- return 0;
+ if (atomic_inc_return(&t->ldt_device_nr) == 1 &&
+ t->ldt_ops->ldto_start != NULL)
+ t->ldt_ops->ldto_start(t);
+
+ memset(d, 0, sizeof *d);
+ d->ld_type = t;
+ lu_ref_init(&d->ld_reference);
+ INIT_LIST_HEAD(&d->ld_linkage);
+
+ return 0;
}
EXPORT_SYMBOL(lu_device_init);
*/
void lu_device_fini(struct lu_device *d)
{
- struct lu_device_type *t;
+ struct lu_device_type *t = d->ld_type;
- t = d->ld_type;
- if (d->ld_obd != NULL) {
- d->ld_obd->obd_lu_dev = NULL;
- d->ld_obd = NULL;
- }
+ if (d->ld_obd != NULL) {
+ d->ld_obd->obd_lu_dev = NULL;
+ d->ld_obd = NULL;
+ }
- lu_ref_fini(&d->ld_reference);
- LASSERTF(cfs_atomic_read(&d->ld_ref) == 0,
- "Refcount is %u\n", cfs_atomic_read(&d->ld_ref));
- LASSERT(t->ldt_device_nr > 0);
- if (--t->ldt_device_nr == 0 && t->ldt_ops->ldto_stop != NULL)
- t->ldt_ops->ldto_stop(t);
+ lu_ref_fini(&d->ld_reference);
+ LASSERTF(atomic_read(&d->ld_ref) == 0,
+ "Refcount is %u\n", atomic_read(&d->ld_ref));
+ LASSERT(atomic_read(&t->ldt_device_nr) > 0);
+
+ if (atomic_dec_and_test(&t->ldt_device_nr) &&
+ t->ldt_ops->ldto_stop != NULL)
+ t->ldt_ops->ldto_stop(t);
}
EXPORT_SYMBOL(lu_device_fini);
* Initialize object \a o that is part of compound object \a h and was created
* by device \a d.
*/
-int lu_object_init(struct lu_object *o,
- struct lu_object_header *h, struct lu_device *d)
+int lu_object_init(struct lu_object *o, struct lu_object_header *h,
+ struct lu_device *d)
{
- memset(o, 0, sizeof *o);
- o->lo_header = h;
- o->lo_dev = d;
- lu_device_get(d);
- o->lo_dev_ref = lu_ref_add(&d->ld_reference, "lu_object", o);
- CFS_INIT_LIST_HEAD(&o->lo_linkage);
- return 0;
+ memset(o, 0, sizeof(*o));
+ o->lo_header = h;
+ o->lo_dev = d;
+ lu_device_get(d);
+ lu_ref_add_at(&d->ld_reference, &o->lo_dev_ref, "lu_object", o);
+ INIT_LIST_HEAD(&o->lo_linkage);
+
+ return 0;
}
EXPORT_SYMBOL(lu_object_init);
*/
void lu_object_fini(struct lu_object *o)
{
- struct lu_device *dev = o->lo_dev;
+ struct lu_device *dev = o->lo_dev;
- LASSERT(cfs_list_empty(&o->lo_linkage));
+ LASSERT(list_empty(&o->lo_linkage));
- if (dev != NULL) {
- lu_ref_del_at(&dev->ld_reference,
- o->lo_dev_ref , "lu_object", o);
- lu_device_put(dev);
- o->lo_dev = NULL;
- }
+ if (dev != NULL) {
+ lu_ref_del_at(&dev->ld_reference, &o->lo_dev_ref,
+ "lu_object", o);
+ lu_device_put(dev);
+ o->lo_dev = NULL;
+ }
}
EXPORT_SYMBOL(lu_object_fini);
*/
void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
{
- cfs_list_move(&o->lo_linkage, &h->loh_layers);
+ list_move(&o->lo_linkage, &h->loh_layers);
}
EXPORT_SYMBOL(lu_object_add_top);
*/
void lu_object_add(struct lu_object *before, struct lu_object *o)
{
- cfs_list_move(&o->lo_linkage, &before->lo_linkage);
+ list_move(&o->lo_linkage, &before->lo_linkage);
}
EXPORT_SYMBOL(lu_object_add);
int lu_object_header_init(struct lu_object_header *h)
{
memset(h, 0, sizeof *h);
- cfs_atomic_set(&h->loh_ref, 1);
- CFS_INIT_HLIST_NODE(&h->loh_hash);
- CFS_INIT_LIST_HEAD(&h->loh_lru);
- CFS_INIT_LIST_HEAD(&h->loh_layers);
+ atomic_set(&h->loh_ref, 1);
+ INIT_HLIST_NODE(&h->loh_hash);
+ INIT_LIST_HEAD(&h->loh_lru);
+ INIT_LIST_HEAD(&h->loh_layers);
lu_ref_init(&h->loh_reference);
return 0;
}
*/
void lu_object_header_fini(struct lu_object_header *h)
{
- LASSERT(cfs_list_empty(&h->loh_layers));
- LASSERT(cfs_list_empty(&h->loh_lru));
- LASSERT(cfs_hlist_unhashed(&h->loh_hash));
+ LASSERT(list_empty(&h->loh_layers));
+ LASSERT(list_empty(&h->loh_lru));
+ LASSERT(hlist_unhashed(&h->loh_hash));
lu_ref_fini(&h->loh_reference);
}
EXPORT_SYMBOL(lu_object_header_fini);
struct lu_object *lu_object_locate(struct lu_object_header *h,
const struct lu_device_type *dtype)
{
- struct lu_object *o;
+ struct lu_object *o;
- cfs_list_for_each_entry(o, &h->loh_layers, lo_linkage) {
- if (o->lo_dev->ld_type == dtype)
- return o;
- }
- return NULL;
+ list_for_each_entry(o, &h->loh_layers, lo_linkage) {
+ if (o->lo_dev->ld_type == dtype)
+ return o;
+ }
+ return NULL;
}
EXPORT_SYMBOL(lu_object_locate);
-
-
/**
* Finalize and free devices in the device stack.
*
/* purge again. */
lu_site_purge(env, site, ~0);
- if (!cfs_hash_is_empty(site->ls_obj_hash)) {
- /*
- * Uh-oh, objects still exist.
- */
- static DECLARE_LU_CDEBUG_PRINT_INFO(cookie, D_ERROR);
-
- lu_site_print(env, site, &cookie, lu_cdebug_printer);
- }
-
for (scan = top; scan != NULL; scan = next) {
const struct lu_device_type *ldt = scan->ld_type;
struct obd_type *type;
}
}
}
-EXPORT_SYMBOL(lu_stack_fini);
enum {
/**
* Maximal number of tld slots.
*/
- LU_CONTEXT_KEY_NR = 32
+ LU_CONTEXT_KEY_NR = 40
};
static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
-static cfs_spinlock_t lu_keys_guard = CFS_SPIN_LOCK_UNLOCKED;
+static DEFINE_SPINLOCK(lu_keys_guard);
+static atomic_t lu_key_initing_cnt = ATOMIC_INIT(0);
/**
* Global counter incremented whenever key is registered, unregistered,
*/
int lu_context_key_register(struct lu_context_key *key)
{
- int result;
- int i;
+ int result;
+ unsigned int i;
LASSERT(key->lct_init != NULL);
LASSERT(key->lct_fini != NULL);
LASSERT(key->lct_owner != NULL);
result = -ENFILE;
- cfs_spin_lock(&lu_keys_guard);
+ spin_lock(&lu_keys_guard);
for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
if (lu_keys[i] == NULL) {
key->lct_index = i;
- cfs_atomic_set(&key->lct_used, 1);
+ atomic_set(&key->lct_used, 1);
lu_keys[i] = key;
lu_ref_init(&key->lct_reference);
result = 0;
break;
}
}
- cfs_spin_unlock(&lu_keys_guard);
- return result;
+ spin_unlock(&lu_keys_guard);
+ return result;
}
EXPORT_SYMBOL(lu_context_key_register);
key = lu_keys[index];
LASSERT(key != NULL);
LASSERT(key->lct_fini != NULL);
- LASSERT(cfs_atomic_read(&key->lct_used) > 1);
+ LASSERT(atomic_read(&key->lct_used) > 1);
key->lct_fini(ctx, key, ctx->lc_value[index]);
lu_ref_del(&key->lct_reference, "ctx", ctx);
- cfs_atomic_dec(&key->lct_used);
- LASSERT(key->lct_owner != NULL);
- if (!(ctx->lc_tags & LCT_NOREF)) {
- LASSERT(cfs_module_refcount(key->lct_owner) > 0);
- cfs_module_put(key->lct_owner);
- }
- ctx->lc_value[index] = NULL;
- }
+ atomic_dec(&key->lct_used);
+
+ LASSERT(key->lct_owner != NULL);
+ if ((ctx->lc_tags & LCT_NOREF) == 0) {
+ LINVRNT(module_refcount(key->lct_owner) > 0);
+ module_put(key->lct_owner);
+ }
+ ctx->lc_value[index] = NULL;
+ }
}
/**
*/
void lu_context_key_degister(struct lu_context_key *key)
{
- LASSERT(cfs_atomic_read(&key->lct_used) >= 1);
- LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
-
- lu_context_key_quiesce(key);
-
- ++key_set_version;
- cfs_spin_lock(&lu_keys_guard);
- key_fini(&lu_shrink_env.le_ctx, key->lct_index);
- if (lu_keys[key->lct_index]) {
- lu_keys[key->lct_index] = NULL;
- lu_ref_fini(&key->lct_reference);
- }
- cfs_spin_unlock(&lu_keys_guard);
-
- LASSERTF(cfs_atomic_read(&key->lct_used) == 1,
- "key has instances: %d\n",
- cfs_atomic_read(&key->lct_used));
+ LASSERT(atomic_read(&key->lct_used) >= 1);
+ LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
+
+ lu_context_key_quiesce(key);
+
+ ++key_set_version;
+ spin_lock(&lu_keys_guard);
+ key_fini(&lu_shrink_env.le_ctx, key->lct_index);
+
+ /**
+ * Wait until all transient contexts referencing this key have
+ * run lu_context_key::lct_fini() method.
+ */
+ while (atomic_read(&key->lct_used) > 1) {
+ spin_unlock(&lu_keys_guard);
+ CDEBUG(D_INFO, "lu_context_key_degister: \"%s\" %p, %d\n",
+ key->lct_owner ? key->lct_owner->name : "", key,
+ atomic_read(&key->lct_used));
+ schedule();
+ spin_lock(&lu_keys_guard);
+ }
+ if (lu_keys[key->lct_index]) {
+ lu_keys[key->lct_index] = NULL;
+ lu_ref_fini(&key->lct_reference);
+ }
+ spin_unlock(&lu_keys_guard);
+
+ LASSERTF(atomic_read(&key->lct_used) == 1,
+ "key has instances: %d\n",
+ atomic_read(&key->lct_used));
}
EXPORT_SYMBOL(lu_context_key_degister);
/**
* List of remembered contexts. XXX document me.
*/
-static CFS_LIST_HEAD(lu_context_remembered);
+static struct list_head lu_context_remembered;
/**
* Destroy \a key in all remembered contexts. This is used to destroy key
* XXX layering violation.
*/
cl_env_cache_purge(~0);
- key->lct_tags |= LCT_QUIESCENT;
/*
* XXX memory barrier has to go here.
*/
- cfs_spin_lock(&lu_keys_guard);
- cfs_list_for_each_entry(ctx, &lu_context_remembered,
- lc_remember)
- key_fini(ctx, key->lct_index);
- cfs_spin_unlock(&lu_keys_guard);
- ++key_set_version;
- }
+ spin_lock(&lu_keys_guard);
+ key->lct_tags |= LCT_QUIESCENT;
+
+ /**
+ * Wait until all lu_context_key::lct_init() methods
+ * have completed.
+ */
+ while (atomic_read(&lu_key_initing_cnt) > 0) {
+ spin_unlock(&lu_keys_guard);
+ CDEBUG(D_INFO, "lu_context_key_quiesce: \"%s\""
+ " %p, %d (%d)\n",
+ key->lct_owner ? key->lct_owner->name : "",
+ key, atomic_read(&key->lct_used),
+ atomic_read(&lu_key_initing_cnt));
+ schedule();
+ spin_lock(&lu_keys_guard);
+ }
+
+ list_for_each_entry(ctx, &lu_context_remembered,
+ lc_remember)
+ key_fini(ctx, key->lct_index);
+ spin_unlock(&lu_keys_guard);
+ ++key_set_version;
+ }
}
-EXPORT_SYMBOL(lu_context_key_quiesce);
void lu_context_key_revive(struct lu_context_key *key)
{
key->lct_tags &= ~LCT_QUIESCENT;
++key_set_version;
}
-EXPORT_SYMBOL(lu_context_key_revive);
static void keys_fini(struct lu_context *ctx)
{
- int i;
+ unsigned int i;
- cfs_spin_lock(&lu_keys_guard);
- if (ctx->lc_value != NULL) {
- for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
- key_fini(ctx, i);
- OBD_FREE(ctx->lc_value,
- ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
- ctx->lc_value = NULL;
- }
- cfs_spin_unlock(&lu_keys_guard);
+ if (ctx->lc_value == NULL)
+ return;
+
+ for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
+ key_fini(ctx, i);
+
+ OBD_FREE(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
+ ctx->lc_value = NULL;
}
static int keys_fill(struct lu_context *ctx)
{
- int i;
+ unsigned int i;
+
+ /*
+ * A serialisation with lu_context_key_quiesce() is needed, but some
+ * "key->lct_init()" are calling kernel memory allocation routine and
+ * can't be called while holding a spin_lock.
+ * "lu_keys_guard" is held while incrementing "lu_key_initing_cnt"
+ * to ensure the start of the serialisation.
+ * An atomic_t variable is still used, in order not to reacquire the
+ * lock when decrementing the counter.
+ */
+ spin_lock(&lu_keys_guard);
+ atomic_inc(&lu_key_initing_cnt);
+ spin_unlock(&lu_keys_guard);
+ LINVRNT(ctx->lc_value != NULL);
for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
struct lu_context_key *key;
LINVRNT(key->lct_init != NULL);
LINVRNT(key->lct_index == i);
- value = key->lct_init(ctx, key);
- if (unlikely(IS_ERR(value)))
- return PTR_ERR(value);
-
- LASSERT(key->lct_owner != NULL);
- if (!(ctx->lc_tags & LCT_NOREF))
- cfs_try_module_get(key->lct_owner);
- lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
- cfs_atomic_inc(&key->lct_used);
+ LASSERT(key->lct_owner != NULL);
+ if (!(ctx->lc_tags & LCT_NOREF) &&
+ try_module_get(key->lct_owner) == 0) {
+ /* module is unloading, skip this key */
+ continue;
+ }
+
+ value = key->lct_init(ctx, key);
+ if (unlikely(IS_ERR(value))) {
+ atomic_dec(&lu_key_initing_cnt);
+ return PTR_ERR(value);
+ }
+
+ lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
+ atomic_inc(&key->lct_used);
/*
* This is the only place in the code, where an
* element of ctx->lc_value[] array is set to non-NULL
}
ctx->lc_version = key_set_version;
}
+ atomic_dec(&lu_key_initing_cnt);
return 0;
}
static int keys_init(struct lu_context *ctx)
{
- int result;
+ OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
+ if (likely(ctx->lc_value != NULL))
+ return keys_fill(ctx);
- OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
- if (likely(ctx->lc_value != NULL))
- result = keys_fill(ctx);
- else
- result = -ENOMEM;
-
- if (result != 0)
- keys_fini(ctx);
- return result;
+ return -ENOMEM;
}
/**
*/
int lu_context_init(struct lu_context *ctx, __u32 tags)
{
- memset(ctx, 0, sizeof *ctx);
- ctx->lc_state = LCS_INITIALIZED;
- ctx->lc_tags = tags;
- if (tags & LCT_REMEMBER) {
- cfs_spin_lock(&lu_keys_guard);
- cfs_list_add(&ctx->lc_remember, &lu_context_remembered);
- cfs_spin_unlock(&lu_keys_guard);
- } else
- CFS_INIT_LIST_HEAD(&ctx->lc_remember);
- return keys_init(ctx);
+ int rc;
+
+ memset(ctx, 0, sizeof *ctx);
+ ctx->lc_state = LCS_INITIALIZED;
+ ctx->lc_tags = tags;
+ if (tags & LCT_REMEMBER) {
+ spin_lock(&lu_keys_guard);
+ list_add(&ctx->lc_remember, &lu_context_remembered);
+ spin_unlock(&lu_keys_guard);
+ } else {
+ INIT_LIST_HEAD(&ctx->lc_remember);
+ }
+
+ rc = keys_init(ctx);
+ if (rc != 0)
+ lu_context_fini(ctx);
+
+ return rc;
}
EXPORT_SYMBOL(lu_context_init);
*/
void lu_context_fini(struct lu_context *ctx)
{
- LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
- ctx->lc_state = LCS_FINALIZED;
- keys_fini(ctx);
- cfs_spin_lock(&lu_keys_guard);
- cfs_list_del_init(&ctx->lc_remember);
- cfs_spin_unlock(&lu_keys_guard);
+ LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
+ ctx->lc_state = LCS_FINALIZED;
+
+ if ((ctx->lc_tags & LCT_REMEMBER) == 0) {
+ LASSERT(list_empty(&ctx->lc_remember));
+ keys_fini(ctx);
+
+ } else { /* could race with key degister */
+ spin_lock(&lu_keys_guard);
+ keys_fini(ctx);
+ list_del_init(&ctx->lc_remember);
+ spin_unlock(&lu_keys_guard);
+ }
}
EXPORT_SYMBOL(lu_context_fini);
*/
void lu_context_exit(struct lu_context *ctx)
{
- int i;
+ unsigned int i;
LINVRNT(ctx->lc_state == LCS_ENTERED);
ctx->lc_state = LCS_LEFT;
if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
- if (ctx->lc_value[i] != NULL) {
- struct lu_context_key *key;
-
- key = lu_keys[i];
- LASSERT(key != NULL);
- if (key->lct_exit != NULL)
- key->lct_exit(ctx,
- key, ctx->lc_value[i]);
- }
+ /* could race with key quiescency */
+ if (ctx->lc_tags & LCT_REMEMBER)
+ spin_lock(&lu_keys_guard);
+ if (ctx->lc_value[i] != NULL) {
+ struct lu_context_key *key;
+
+ key = lu_keys[i];
+ LASSERT(key != NULL);
+ if (key->lct_exit != NULL)
+ key->lct_exit(ctx,
+ key, ctx->lc_value[i]);
+ }
+ if (ctx->lc_tags & LCT_REMEMBER)
+ spin_unlock(&lu_keys_guard);
}
}
}
/**
* Allocate for context all missing keys that were registered after context
- * creation.
+ * creation. key_set_version is only changed in rare cases when modules
+ * are loaded and removed.
*/
int lu_context_refill(struct lu_context *ctx)
{
- LINVRNT(ctx->lc_value != NULL);
- return ctx->lc_version == key_set_version ? 0 : keys_fill(ctx);
+ return likely(ctx->lc_version == key_set_version) ? 0 : keys_fill(ctx);
+}
+
+/**
+ * lu_ctx_tags/lu_ses_tags will be updated if there are new types of
+ * obd being added. Currently, this is only used on client side, specifically
+ * for echo device client, for other stack (like ptlrpc threads), context are
+ * predefined when the lu_device type are registered, during the module probe
+ * phase.
+ */
+__u32 lu_context_tags_default = 0;
+__u32 lu_session_tags_default = 0;
+
+void lu_context_tags_update(__u32 tags)
+{
+ spin_lock(&lu_keys_guard);
+ lu_context_tags_default |= tags;
+ key_set_version++;
+ spin_unlock(&lu_keys_guard);
}
-EXPORT_SYMBOL(lu_context_refill);
+EXPORT_SYMBOL(lu_context_tags_update);
+
+void lu_context_tags_clear(__u32 tags)
+{
+ spin_lock(&lu_keys_guard);
+ lu_context_tags_default &= ~tags;
+ key_set_version++;
+ spin_unlock(&lu_keys_guard);
+}
+EXPORT_SYMBOL(lu_context_tags_clear);
+
+void lu_session_tags_update(__u32 tags)
+{
+ spin_lock(&lu_keys_guard);
+ lu_session_tags_default |= tags;
+ key_set_version++;
+ spin_unlock(&lu_keys_guard);
+}
+EXPORT_SYMBOL(lu_session_tags_update);
+
+void lu_session_tags_clear(__u32 tags)
+{
+ spin_lock(&lu_keys_guard);
+ lu_session_tags_default &= ~tags;
+ key_set_version++;
+ spin_unlock(&lu_keys_guard);
+}
+EXPORT_SYMBOL(lu_session_tags_clear);
int lu_env_init(struct lu_env *env, __u32 tags)
{
}
EXPORT_SYMBOL(lu_env_refill);
-static struct cfs_shrinker *lu_site_shrinker = NULL;
+/**
+ * Currently, this API will only be used by echo client.
+ * Because echo client and normal lustre client will share
+ * same cl_env cache. So echo client needs to refresh
+ * the env context after it get one from the cache, especially
+ * when normal client and echo client co-exist in the same client.
+ */
+int lu_env_refill_by_tags(struct lu_env *env, __u32 ctags,
+ __u32 stags)
+{
+ int result;
+
+ if ((env->le_ctx.lc_tags & ctags) != ctags) {
+ env->le_ctx.lc_version = 0;
+ env->le_ctx.lc_tags |= ctags;
+ }
+
+ if (env->le_ses && (env->le_ses->lc_tags & stags) != stags) {
+ env->le_ses->lc_version = 0;
+ env->le_ses->lc_tags |= stags;
+ }
+
+ result = lu_env_refill(env);
+
+ return result;
+}
+EXPORT_SYMBOL(lu_env_refill_by_tags);
+
+static struct shrinker *lu_site_shrinker;
typedef struct lu_site_stats{
unsigned lss_populated;
static void lu_site_stats_get(cfs_hash_t *hs,
lu_site_stats_t *stats, int populated)
{
- cfs_hash_bd_t bd;
- int i;
+ cfs_hash_bd_t bd;
+ unsigned int i;
cfs_hash_for_each_bucket(hs, &bd, i) {
struct lu_site_bkt_data *bkt = cfs_hash_bd_extra_get(hs, &bd);
- cfs_hlist_head_t *hhead;
+ struct hlist_head *hhead;
cfs_hash_bd_lock(hs, &bd, 1);
- stats->lss_busy += bkt->lsb_busy;
+ stats->lss_busy +=
+ cfs_hash_bd_count_get(&bd) - bkt->lsb_lru_len;
stats->lss_total += cfs_hash_bd_count_get(&bd);
stats->lss_max_search = max((int)stats->lss_max_search,
cfs_hash_bd_depmax_get(&bd));
}
cfs_hash_bd_for_each_hlist(hs, &bd, hhead) {
- if (!cfs_hlist_empty(hhead))
+ if (!hlist_empty(hhead))
stats->lss_populated++;
}
cfs_hash_bd_unlock(hs, &bd, 1);
}
}
-#ifdef __KERNEL__
-static int lu_cache_shrink(SHRINKER_FIRST_ARG int nr_to_scan,
- unsigned int gfp_mask)
+
+static unsigned long lu_cache_shrink_count(struct shrinker *sk,
+ struct shrink_control *sc)
+{
+ lu_site_stats_t stats;
+ struct lu_site *s;
+ struct lu_site *tmp;
+ unsigned long cached = 0;
+
+ if (!(sc->gfp_mask & __GFP_FS))
+ return 0;
+
+ mutex_lock(&lu_sites_guard);
+ list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
+ memset(&stats, 0, sizeof(stats));
+ lu_site_stats_get(s->ls_obj_hash, &stats, 0);
+ cached += stats.lss_total - stats.lss_busy;
+ }
+ mutex_unlock(&lu_sites_guard);
+
+ cached = (cached / 100) * sysctl_vfs_cache_pressure;
+ CDEBUG(D_INODE, "%ld objects cached\n", cached);
+ return cached;
+}
+
+static unsigned long lu_cache_shrink_scan(struct shrinker *sk,
+ struct shrink_control *sc)
+{
+ struct lu_site *s;
+ struct lu_site *tmp;
+ unsigned long remain = sc->nr_to_scan;
+ LIST_HEAD(splice);
+
+ if (!(sc->gfp_mask & __GFP_FS))
+ /* We must not take the lu_sites_guard lock when
+ * __GFP_FS is *not* set because of the deadlock
+ * possibility detailed above. Additionally,
+ * since we cannot determine the number of
+ * objects in the cache without taking this
+ * lock, we're in a particularly tough spot. As
+ * a result, we'll just lie and say our cache is
+ * empty. This _should_ be ok, as we can't
+ * reclaim objects when __GFP_FS is *not* set
+ * anyways.
+ */
+ return SHRINK_STOP;
+
+ mutex_lock(&lu_sites_guard);
+ list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
+ remain = lu_site_purge(&lu_shrink_env, s, remain);
+ /*
+ * Move just shrunk site to the tail of site list to
+ * assure shrinking fairness.
+ */
+ list_move_tail(&s->ls_linkage, &splice);
+ }
+ list_splice(&splice, lu_sites.prev);
+ mutex_unlock(&lu_sites_guard);
+
+ return sc->nr_to_scan - remain;
+}
+
+#ifndef HAVE_SHRINKER_COUNT
+/*
+ * There exists a potential lock inversion deadlock scenario when using
+ * Lustre on top of ZFS. This occurs between one of ZFS's
+ * buf_hash_table.ht_lock's, and Lustre's lu_sites_guard lock. Essentially,
+ * thread A will take the lu_sites_guard lock and sleep on the ht_lock,
+ * while thread B will take the ht_lock and sleep on the lu_sites_guard
+ * lock. Obviously neither thread will wake and drop their respective hold
+ * on their lock.
+ *
+ * To prevent this from happening we must ensure the lu_sites_guard lock is
+ * not taken while down this code path. ZFS reliably does not set the
+ * __GFP_FS bit in its code paths, so this can be used to determine if it
+ * is safe to take the lu_sites_guard lock.
+ *
+ * Ideally we should accurately return the remaining number of cached
+ * objects without taking the lu_sites_guard lock, but this is not
+ * possible in the current implementation.
+ */
+static int lu_cache_shrink(SHRINKER_ARGS(sc, nr_to_scan, gfp_mask))
{
- lu_site_stats_t stats;
- struct lu_site *s;
- struct lu_site *tmp;
int cached = 0;
- int remain = nr_to_scan;
- CFS_LIST_HEAD(splice);
+ struct shrink_control scv = {
+ .nr_to_scan = shrink_param(sc, nr_to_scan),
+ .gfp_mask = shrink_param(sc, gfp_mask)
+ };
+#if !defined(HAVE_SHRINKER_WANT_SHRINK_PTR) && !defined(HAVE_SHRINK_CONTROL)
+ struct shrinker* shrinker = NULL;
+#endif
- if (nr_to_scan != 0) {
- if (!(gfp_mask & __GFP_FS))
- return -1;
- CDEBUG(D_INODE, "Shrink %d objects\n", nr_to_scan);
- }
- cfs_down(&lu_sites_guard);
- cfs_list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
- if (nr_to_scan != 0) {
- remain = lu_site_purge(&lu_shrink_env, s, remain);
- /*
- * Move just shrunk site to the tail of site list to
- * assure shrinking fairness.
- */
- cfs_list_move_tail(&s->ls_linkage, &splice);
- }
+ CDEBUG(D_INODE, "Shrink %lu objects\n", scv.nr_to_scan);
- memset(&stats, 0, sizeof(stats));
- lu_site_stats_get(s->ls_obj_hash, &stats, 0);
- cached += stats.lss_total - stats.lss_busy;
- if (nr_to_scan && remain <= 0)
- break;
- }
- cfs_list_splice(&splice, lu_sites.prev);
- cfs_up(&lu_sites_guard);
+ lu_cache_shrink_scan(shrinker, &scv);
- cached = (cached / 100) * sysctl_vfs_cache_pressure;
- if (nr_to_scan == 0)
- CDEBUG(D_INODE, "%d objects cached\n", cached);
- return cached;
+ cached = lu_cache_shrink_count(shrinker, &scv);
+ if (scv.nr_to_scan == 0)
+ CDEBUG(D_INODE, "%d objects cached\n", cached);
+ return cached;
}
+#endif /* HAVE_SHRINKER_COUNT */
+
+
/*
* Debugging stuff.
*/
/**
* Environment to be used in debugger, contains all tags.
*/
-struct lu_env lu_debugging_env;
+static struct lu_env lu_debugging_env;
/**
* Debugging printer function using printk().
*/
int lu_printk_printer(const struct lu_env *env,
- void *unused, const char *format, ...)
+ void *unused, const char *format, ...)
{
va_list args;
return 0;
}
-void lu_debugging_setup(void)
+int lu_debugging_setup(void)
{
- lu_env_init(&lu_debugging_env, ~0);
+ return lu_env_init(&lu_debugging_env, ~0);
}
void lu_context_keys_dump(void)
{
- int i;
+ unsigned int i;
for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
struct lu_context_key *key;
CERROR("[%d]: %p %x (%p,%p,%p) %d %d \"%s\"@%p\n",
i, key, key->lct_tags,
key->lct_init, key->lct_fini, key->lct_exit,
- key->lct_index, cfs_atomic_read(&key->lct_used),
+ key->lct_index, atomic_read(&key->lct_used),
key->lct_owner ? key->lct_owner->name : "",
key->lct_owner);
lu_ref_print(&key->lct_reference);
}
}
}
-EXPORT_SYMBOL(lu_context_keys_dump);
-#else /* !__KERNEL__ */
-static int lu_cache_shrink(int nr, unsigned int gfp_mask)
-{
- return 0;
-}
-#endif /* __KERNEL__ */
-
-int cl_global_init(void);
-void cl_global_fini(void);
-int lu_ref_global_init(void);
-void lu_ref_global_fini(void);
-
-int dt_global_init(void);
-void dt_global_fini(void);
-
-int llo_global_init(void);
-void llo_global_fini(void);
/**
* Initialization of global lu_* data.
int lu_global_init(void)
{
int result;
+ DEF_SHRINKER_VAR(shvar, lu_cache_shrink,
+ lu_cache_shrink_count, lu_cache_shrink_scan);
CDEBUG(D_INFO, "Lustre LU module (%p).\n", &lu_keys);
+ INIT_LIST_HEAD(&lu_device_types);
+ INIT_LIST_HEAD(&lu_context_remembered);
+ INIT_LIST_HEAD(&lu_sites);
+
result = lu_ref_global_init();
if (result != 0)
return result;
result = lu_context_key_register(&lu_global_key);
if (result != 0)
return result;
+
/*
* At this level, we don't know what tags are needed, so allocate them
* conservatively. This should not be too bad, because this
* environment is global.
*/
- cfs_down(&lu_sites_guard);
+ mutex_lock(&lu_sites_guard);
result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
- cfs_up(&lu_sites_guard);
+ mutex_unlock(&lu_sites_guard);
if (result != 0)
return result;
* inode, one for ea. Unfortunately setting this high value results in
* lu_object/inode cache consuming all the memory.
*/
- lu_site_shrinker = cfs_set_shrinker(CFS_DEFAULT_SEEKS, lu_cache_shrink);
+ lu_site_shrinker = set_shrinker(DEFAULT_SEEKS, &shvar);
if (lu_site_shrinker == NULL)
return -ENOMEM;
- result = lu_time_global_init();
- if (result)
- GOTO(out, result);
-
-#ifdef __KERNEL__
- result = dt_global_init();
- if (result)
- GOTO(out, result);
-
- result = llo_global_init();
- if (result)
- GOTO(out, result);
-#endif
- result = cl_global_init();
-out:
-
return result;
}
*/
void lu_global_fini(void)
{
- cl_global_fini();
-#ifdef __KERNEL__
- llo_global_fini();
- dt_global_fini();
-#endif
- lu_time_global_fini();
if (lu_site_shrinker != NULL) {
- cfs_remove_shrinker(lu_site_shrinker);
+ remove_shrinker(lu_site_shrinker);
lu_site_shrinker = NULL;
}
- lu_context_key_degister(&lu_global_key);
+ lu_context_key_degister(&lu_global_key);
/*
* Tear shrinker environment down _after_ de-registering
* lu_global_key, because the latter has a value in the former.
*/
- cfs_down(&lu_sites_guard);
+ mutex_lock(&lu_sites_guard);
lu_env_fini(&lu_shrink_env);
- cfs_up(&lu_sites_guard);
+ mutex_unlock(&lu_sites_guard);
lu_ref_global_fini();
}
-struct lu_buf LU_BUF_NULL = {
- .lb_buf = NULL,
- .lb_len = 0
-};
-EXPORT_SYMBOL(LU_BUF_NULL);
-
static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx)
{
-#ifdef LPROCFS
+#ifdef CONFIG_PROC_FS
struct lprocfs_counter ret;
lprocfs_stats_collect(stats, idx, &ret);
* Output site statistical counters into a buffer. Suitable for
* lprocfs_rd_*()-style functions.
*/
+int lu_site_stats_seq_print(const struct lu_site *s, struct seq_file *m)
+{
+ lu_site_stats_t stats;
+
+ memset(&stats, 0, sizeof(stats));
+ lu_site_stats_get(s->ls_obj_hash, &stats, 1);
+
+ return seq_printf(m, "%d/%d %d/%d %d %d %d %d %d %d %d\n",
+ stats.lss_busy,
+ stats.lss_total,
+ stats.lss_populated,
+ CFS_HASH_NHLIST(s->ls_obj_hash),
+ stats.lss_max_search,
+ ls_stats_read(s->ls_stats, LU_SS_CREATED),
+ ls_stats_read(s->ls_stats, LU_SS_CACHE_HIT),
+ ls_stats_read(s->ls_stats, LU_SS_CACHE_MISS),
+ ls_stats_read(s->ls_stats, LU_SS_CACHE_RACE),
+ ls_stats_read(s->ls_stats, LU_SS_CACHE_DEATH_RACE),
+ ls_stats_read(s->ls_stats, LU_SS_LRU_PURGED));
+}
+EXPORT_SYMBOL(lu_site_stats_seq_print);
+
int lu_site_stats_print(const struct lu_site *s, char *page, int count)
{
lu_site_stats_t stats;
ls_stats_read(s->ls_stats, LU_SS_CACHE_DEATH_RACE),
ls_stats_read(s->ls_stats, LU_SS_LRU_PURGED));
}
-EXPORT_SYMBOL(lu_site_stats_print);
-
-const char *lu_time_names[LU_TIME_NR] = {
- [LU_TIME_FIND_LOOKUP] = "find_lookup",
- [LU_TIME_FIND_ALLOC] = "find_alloc",
- [LU_TIME_FIND_INSERT] = "find_insert"
-};
-EXPORT_SYMBOL(lu_time_names);
/**
* Helper function to initialize a number of kmem slab caches at once.
int lu_kmem_init(struct lu_kmem_descr *caches)
{
int result;
+ struct lu_kmem_descr *iter = caches;
- for (result = 0; caches->ckd_cache != NULL; ++caches) {
- *caches->ckd_cache = cfs_mem_cache_create(caches->ckd_name,
- caches->ckd_size,
- 0, 0);
- if (*caches->ckd_cache == NULL) {
+ for (result = 0; iter->ckd_cache != NULL; ++iter) {
+ *iter->ckd_cache = kmem_cache_create(iter->ckd_name,
+ iter->ckd_size,
+ 0, 0, NULL);
+ if (*iter->ckd_cache == NULL) {
result = -ENOMEM;
+ /* free all previously allocated caches */
+ lu_kmem_fini(caches);
break;
}
}
*/
void lu_kmem_fini(struct lu_kmem_descr *caches)
{
- int rc;
-
for (; caches->ckd_cache != NULL; ++caches) {
if (*caches->ckd_cache != NULL) {
- rc = cfs_mem_cache_destroy(*caches->ckd_cache);
- LASSERTF(rc == 0, "couldn't destroy %s slab\n",
- caches->ckd_name);
+ kmem_cache_destroy(*caches->ckd_cache);
*caches->ckd_cache = NULL;
}
}
}
EXPORT_SYMBOL(lu_kmem_fini);
+
+/**
+ * Temporary solution to be able to assign fid in ->do_create()
+ * till we have fully-functional OST fids
+ */
+void lu_object_assign_fid(const struct lu_env *env, struct lu_object *o,
+ const struct lu_fid *fid)
+{
+ struct lu_site *s = o->lo_dev->ld_site;
+ struct lu_fid *old = &o->lo_header->loh_fid;
+ struct lu_object *shadow;
+ wait_queue_t waiter;
+ cfs_hash_t *hs;
+ cfs_hash_bd_t bd;
+ __u64 version = 0;
+
+ LASSERT(fid_is_zero(old));
+
+ hs = s->ls_obj_hash;
+ cfs_hash_bd_get_and_lock(hs, (void *)fid, &bd, 1);
+ shadow = htable_lookup(s, &bd, fid, &waiter, &version);
+ /* supposed to be unique */
+ LASSERT(IS_ERR(shadow) && PTR_ERR(shadow) == -ENOENT);
+ *old = *fid;
+ cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
+ cfs_hash_bd_unlock(hs, &bd, 1);
+}
+EXPORT_SYMBOL(lu_object_assign_fid);
+
+/**
+ * allocates object with 0 (non-assiged) fid
+ * XXX: temporary solution to be able to assign fid in ->do_create()
+ * till we have fully-functional OST fids
+ */
+struct lu_object *lu_object_anon(const struct lu_env *env,
+ struct lu_device *dev,
+ const struct lu_object_conf *conf)
+{
+ struct lu_fid fid;
+ struct lu_object *o;
+
+ fid_zero(&fid);
+ o = lu_object_alloc(env, dev, &fid, conf);
+
+ return o;
+}
+EXPORT_SYMBOL(lu_object_anon);
+
+struct lu_buf LU_BUF_NULL = {
+ .lb_buf = NULL,
+ .lb_len = 0
+};
+EXPORT_SYMBOL(LU_BUF_NULL);
+
+void lu_buf_free(struct lu_buf *buf)
+{
+ LASSERT(buf);
+ if (buf->lb_buf) {
+ LASSERT(buf->lb_len > 0);
+ OBD_FREE_LARGE(buf->lb_buf, buf->lb_len);
+ buf->lb_buf = NULL;
+ buf->lb_len = 0;
+ }
+}
+EXPORT_SYMBOL(lu_buf_free);
+
+void lu_buf_alloc(struct lu_buf *buf, size_t size)
+{
+ LASSERT(buf);
+ LASSERT(buf->lb_buf == NULL);
+ LASSERT(buf->lb_len == 0);
+ OBD_ALLOC_LARGE(buf->lb_buf, size);
+ if (likely(buf->lb_buf))
+ buf->lb_len = size;
+}
+EXPORT_SYMBOL(lu_buf_alloc);
+
+void lu_buf_realloc(struct lu_buf *buf, size_t size)
+{
+ lu_buf_free(buf);
+ lu_buf_alloc(buf, size);
+}
+EXPORT_SYMBOL(lu_buf_realloc);
+
+struct lu_buf *lu_buf_check_and_alloc(struct lu_buf *buf, size_t len)
+{
+ if (buf->lb_buf == NULL && buf->lb_len == 0)
+ lu_buf_alloc(buf, len);
+
+ if ((len > buf->lb_len) && (buf->lb_buf != NULL))
+ lu_buf_realloc(buf, len);
+
+ return buf;
+}
+EXPORT_SYMBOL(lu_buf_check_and_alloc);
+
+/**
+ * Increase the size of the \a buf.
+ * preserves old data in buffer
+ * old buffer remains unchanged on error
+ * \retval 0 or -ENOMEM
+ */
+int lu_buf_check_and_grow(struct lu_buf *buf, size_t len)
+{
+ char *ptr;
+
+ if (len <= buf->lb_len)
+ return 0;
+
+ OBD_ALLOC_LARGE(ptr, len);
+ if (ptr == NULL)
+ return -ENOMEM;
+
+ /* Free the old buf */
+ if (buf->lb_buf != NULL) {
+ memcpy(ptr, buf->lb_buf, buf->lb_len);
+ OBD_FREE_LARGE(buf->lb_buf, buf->lb_len);
+ }
+
+ buf->lb_buf = ptr;
+ buf->lb_len = len;
+ return 0;
+}