X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fobdclass%2Flu_object.c;h=4fd6f0df649a97e64c0b48a77e7d4f0dcc975c3c;hb=a7a2133bfab42eba077f1b8d5c991c651c8028c3;hp=5077327efe8ee2954c0a1cc2321300db00ba8e8a;hpb=cbfee37be595d9f378e25b81f516839caa3ee7ff;p=fs%2Flustre-release.git diff --git a/lustre/obdclass/lu_object.c b/lustre/obdclass/lu_object.c index 5077327..4fd6f0d 100644 --- a/lustre/obdclass/lu_object.c +++ b/lustre/obdclass/lu_object.c @@ -194,10 +194,10 @@ static struct lu_object *lu_object_alloc(const struct lu_env *env, */ static void lu_object_free(const struct lu_env *env, struct lu_object *o) { - struct list_head splice; + struct list_head splice; struct lu_object *scan; - struct lu_site *site; - struct list_head *layers; + struct lu_site *site; + struct list_head *layers; site = o->lo_dev->ld_site; layers = &o->lo_header->loh_layers; @@ -336,7 +336,7 @@ int lu_cdebug_printer(const struct lu_env *env, struct lu_cdebug_data *key; int used; int complete; - va_list args; + va_list args; va_start(args, format); @@ -351,9 +351,10 @@ int lu_cdebug_printer(const struct lu_env *env, vsnprintf(key->lck_area + used, ARRAY_SIZE(key->lck_area) - used, format, args); if (complete) { - 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 (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); key->lck_area[0] = 0; } va_end(args); @@ -361,23 +362,24 @@ int lu_cdebug_printer(const struct lu_env *env, } EXPORT_SYMBOL(lu_cdebug_printer); -/* +/** * Print object header. */ -static void lu_object_header_print(const struct lu_env *env, - void *cookie, lu_printer_t printer, - const struct lu_object_header *hdr) +void lu_object_header_print(const struct lu_env *env, void *cookie, + 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, atomic_read(&hdr->loh_ref), PFID(&hdr->loh_fid), hlist_unhashed(&hdr->loh_hash) ? "" : " hash", - list_empty(&hdr->loh_lru) ? "" : " lru", + 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 @o to the @printer. +/** + * 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) @@ -388,21 +390,24 @@ void lu_object_print(const struct lu_env *env, void *cookie, top = o->lo_header; lu_object_header_print(env, cookie, printer, top); - (*printer)(env, cookie, "\n"); + (*printer)(env, cookie, "{ \n"); list_for_each_entry(o, &top->loh_layers, lo_linkage) { depth = o->lo_depth + 4; - LASSERT(o->lo_ops->loo_object_print != NULL); + /* - * print `.' @depth times. + * print `.' \a depth times followed by type name and address */ - (*printer)(env, cookie, "%*.*s", depth, depth, ruler); - o->lo_ops->loo_object_print(env, cookie, printer, o); + (*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); -/* +/** * Check object consistency. */ int lu_object_invariant(const struct lu_object *o) @@ -421,15 +426,29 @@ EXPORT_SYMBOL(lu_object_invariant); static struct lu_object *htable_lookup(struct lu_site *s, const struct hlist_head *bucket, - const struct lu_fid *f) + const struct lu_fid *f, + cfs_waitlink_t *waiter) { struct lu_object_header *h; struct hlist_node *scan; hlist_for_each_entry(h, scan, bucket, loh_hash) { s->ls_stats.s_cache_check ++; - if (likely(lu_fid_eq(&h->loh_fid, f) && - !lu_object_is_dying(h))) { + if (likely(lu_fid_eq(&h->loh_fid, f))) { + if (unlikely(lu_object_is_dying(h))) { + /* + * Lookup found an object being destroyed; + * this object cannot be returned (to assure + * that references to dying objects are + * eventually drained), and moreover, lookup + * has to wait until object is freed. + */ + cfs_waitlink_init(waiter); + cfs_waitq_add(&s->ls_marche_funebre, waiter); + set_current_state(CFS_TASK_UNINT); + s->ls_stats.s_cache_death_race ++; + return ERR_PTR(-EAGAIN); + } /* bump reference count... */ if (atomic_add_return(1, &h->loh_ref) == 1) ++ s->ls_busy; @@ -454,16 +473,31 @@ static __u32 fid_hash(const struct lu_fid *f, int bits) return hash_long(fid_flatten(f), bits); } -/* - * Search cache for an object with the fid @f. If such object is found, return - * it. Otherwise, create new object, insert it into cache and return it. In - * any case, additional reference is acquired on the returned object. +/** + * Search cache for an object with the fid \a f. If such object is found, + * return it. Otherwise, create new object, insert it into cache and return + * it. In any case, additional reference is acquired on the returned object. */ struct lu_object *lu_object_find(const struct lu_env *env, - struct lu_site *s, const struct lu_fid *f) + struct lu_device *dev, const struct lu_fid *f, + const struct lu_object_conf *conf) { - struct lu_object *o; - struct lu_object *shadow; + return lu_object_find_at(env, dev->ld_site->ls_top_dev, f, conf); +} +EXPORT_SYMBOL(lu_object_find); + +/** + * 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_site *s; + struct lu_object *o; + struct lu_object *shadow; struct hlist_head *bucket; /* @@ -478,12 +512,16 @@ struct lu_object *lu_object_find(const struct lu_env *env, * object just allocated. * - unlock index; * - return object. + * + * If dying object is found during index search, add @waiter to the + * site wait-queue and return ERR_PTR(-EAGAIN). */ + s = dev->ld_site; bucket = s->ls_hash + fid_hash(f, s->ls_hash_bits); read_lock(&s->ls_guard); - o = htable_lookup(s, bucket, f); + o = htable_lookup(s, bucket, f, waiter); read_unlock(&s->ls_guard); if (o != NULL) @@ -493,14 +531,14 @@ struct lu_object *lu_object_find(const struct lu_env *env, * Allocate new object. This may result in rather complicated * operations, including fld queries, inode loading, etc. */ - o = lu_object_alloc(env, s, f); + o = lu_object_alloc(env, dev, f, conf); if (unlikely(IS_ERR(o))) return o; LASSERT(lu_fid_eq(lu_object_fid(o), f)); write_lock(&s->ls_guard); - shadow = htable_lookup(s, bucket, f); + shadow = htable_lookup(s, bucket, f, waiter); if (likely(shadow == NULL)) { hlist_add_head(&o->lo_header->loh_hash, bucket); list_add_tail(&o->lo_header->loh_lru, &s->ls_lru); @@ -515,9 +553,58 @@ struct lu_object *lu_object_find(const struct lu_env *env, lu_object_free(env, o); return shadow; } -EXPORT_SYMBOL(lu_object_find); -/* +/** + * Much like lu_object_find(), but top level device of object is specifically + * \a dev rather than top level device of the site. This interface allows + * 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_object *obj; + cfs_waitlink_t wait; + + while (1) { + obj = lu_object_find_try(env, dev, f, conf, &wait); + if (obj == ERR_PTR(-EAGAIN)) { + /* + * lu_object_find_try() already added waiter into the + * wait queue. + */ + cfs_waitq_wait(&wait, CFS_TASK_UNINT); + cfs_waitq_del(&dev->ld_site->ls_marche_funebre, &wait); + } else + break; + } + return obj; +} +EXPORT_SYMBOL(lu_object_find_at); + +/** + * Find object with given fid, and return its slice belonging to given device. + */ +struct lu_object *lu_object_find_slice(const struct lu_env *env, + struct lu_device *dev, + const struct lu_fid *f, + const struct lu_object_conf *conf) +{ + struct lu_object *top; + struct lu_object *obj; + + 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; +} +EXPORT_SYMBOL(lu_object_find_slice); + /** * Global list of all device types. */ @@ -757,9 +844,10 @@ void lu_device_fini(struct lu_device *d) struct lu_device_type *t; t = d->ld_type; - if (d->ld_obd != NULL) - /* finish lprocfs */ - lprocfs_obd_cleanup(d->ld_obd); + if (d->ld_obd != NULL) { + d->ld_obd->obd_lu_dev = NULL; + d->ld_obd = NULL; + } lu_ref_fini(&d->ld_reference); LASSERTF(atomic_read(&d->ld_ref) == 0, @@ -877,7 +965,7 @@ EXPORT_SYMBOL(lu_object_locate); /** * Finalize and free devices in the device stack. - * + * * Finalize device stack by purging object cache, and calling * lu_device_type_operations::ldto_device_fini() and * lu_device_type_operations::ldto_device_free() on all devices in the stack. @@ -914,9 +1002,9 @@ void lu_stack_fini(const struct lu_env *env, struct lu_device *top) next = ldt->ldt_ops->ldto_device_free(env, scan); type = ldt->ldt_obd_type; if (type != NULL) { - type->typ_refcnt--; - class_put_type(type); - } + type->typ_refcnt--; + class_put_type(type); + } } } EXPORT_SYMBOL(lu_stack_fini); @@ -973,7 +1061,7 @@ EXPORT_SYMBOL(lu_context_key_register); static void key_fini(struct lu_context *ctx, int index) { - if (ctx->lc_value[index] != NULL) { + if (ctx->lc_value != NULL && ctx->lc_value[index] != NULL) { struct lu_context_key *key; key = lu_keys[index]; @@ -1001,6 +1089,8 @@ void lu_context_key_degister(struct lu_context_key *key) 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; key_fini(&lu_shrink_env.le_ctx, key->lct_index); @@ -1101,6 +1191,7 @@ void *lu_context_key_get(const struct lu_context *ctx, { LINVRNT(ctx->lc_state == LCS_ENTERED); LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys)); + LASSERT(lu_keys[key->lct_index] == key); return ctx->lc_value[key->lct_index]; } EXPORT_SYMBOL(lu_context_key_get); @@ -1118,8 +1209,13 @@ static CFS_LIST_HEAD(lu_context_remembered); void lu_context_key_quiesce(struct lu_context_key *key) { struct lu_context *ctx; + extern unsigned cl_env_cache_purge(unsigned nr); if (!(key->lct_tags & LCT_QUIESCENT)) { + /* + * XXX layering violation. + */ + cl_env_cache_purge(~0); key->lct_tags |= LCT_QUIESCENT; /* * XXX memory barrier has to go here. @@ -1176,6 +1272,7 @@ static int keys_fill(struct lu_context *ctx) 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)) try_module_get(key->lct_owner); @@ -1288,30 +1385,16 @@ int lu_context_refill(struct lu_context *ctx) } EXPORT_SYMBOL(lu_context_refill); -static int lu_env_setup(struct lu_env *env, struct lu_context *ses, - __u32 tags, int noref) +int lu_env_init(struct lu_env *env, __u32 tags) { int result; - LINVRNT(ergo(!noref, !(tags & LCT_NOREF))); - - env->le_ses = ses; + env->le_ses = NULL; result = lu_context_init(&env->le_ctx, tags); if (likely(result == 0)) lu_context_enter(&env->le_ctx); return result; } - -static int lu_env_init_noref(struct lu_env *env, struct lu_context *ses, - __u32 tags) -{ - return lu_env_setup(env, ses, tags, 1); -} - -int lu_env_init(struct lu_env *env, struct lu_context *ses, __u32 tags) -{ - return lu_env_setup(env, ses, tags, 0); -} EXPORT_SYMBOL(lu_env_init); void lu_env_fini(struct lu_env *env) @@ -1368,6 +1451,54 @@ static int lu_cache_shrink(int nr, unsigned int gfp_mask) return cached; } +/* + * Debugging stuff. + */ + +/** + * Environment to be used in debugger, contains all tags. + */ +struct lu_env lu_debugging_env; + +/** + * Debugging printer function using printk(). + */ +int lu_printk_printer(const struct lu_env *env, + void *_, const char *format, ...) +{ + va_list args; + + va_start(args, format); + vprintk(format, args); + va_end(args); + return 0; +} + +void lu_debugging_setup(void) +{ + lu_env_init(&lu_debugging_env, ~0); +} + +void lu_context_keys_dump(void) +{ + int i; + + for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) { + struct lu_context_key *key; + + key = lu_keys[i]; + if (key != NULL) { + CERROR("[%i]: %p %x (%p,%p,%p) %i %i \"%s\"@%p\n", + i, key, key->lct_tags, + key->lct_init, key->lct_fini, key->lct_exit, + 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) { @@ -1375,9 +1506,17 @@ static int lu_cache_shrink(int nr, unsigned int gfp_mask) } #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. */ @@ -1391,21 +1530,21 @@ int lu_global_init(void) 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. - */ - down(&lu_sites_guard); - result = lu_env_init_noref(&lu_shrink_env, NULL, LCT_SHRINKER); - up(&lu_sites_guard); + */ + down(&lu_sites_guard); + result = lu_env_init(&lu_shrink_env, LCT_SHRINKER); + up(&lu_sites_guard); if (result != 0) return result; result = lu_ref_global_init(); if (result != 0) return result; - /* + /* * seeks estimation: 3 seeks to read a record from oi, one to read * inode, one for ea. Unfortunately setting this high value results in * lu_object/inode cache consuming all the memory. @@ -1414,7 +1553,22 @@ int lu_global_init(void) if (lu_site_shrinker == NULL) return -ENOMEM; - result = lu_time_global_init(); + 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; } @@ -1423,6 +1577,11 @@ int lu_global_init(void) */ 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) { remove_shrinker(lu_site_shrinker); @@ -1448,9 +1607,41 @@ struct lu_buf LU_BUF_NULL = { }; EXPORT_SYMBOL(LU_BUF_NULL); +/** + * Output site statistical counters into a buffer. Suitable for + * lprocfs_rd_*()-style functions. + */ +int lu_site_stats_print(const struct lu_site *s, char *page, int count) +{ + int i; + int populated; + + /* + * How many hash buckets are not-empty? Don't bother with locks: it's + * an estimation anyway. + */ + for (i = 0, populated = 0; i < s->ls_hash_size; i++) + populated += !hlist_empty(&s->ls_hash[i]); + + return snprintf(page, count, "%d %d %d/%d %d %d %d %d %d %d %d\n", + s->ls_total, + s->ls_busy, + populated, + s->ls_hash_size, + s->ls_stats.s_created, + s->ls_stats.s_cache_hit, + s->ls_stats.s_cache_miss, + s->ls_stats.s_cache_check, + s->ls_stats.s_cache_race, + s->ls_stats.s_cache_death_race, + s->ls_stats.s_lru_purged); +} +EXPORT_SYMBOL(lu_site_stats_print); + +#ifdef __KERNEL__ /* - * XXX: Functions below logically belong to fid module, but they are used by - * dt_store_open(). Put them here until better place is found. + * XXX: Functions below logically belong to the fid module, but they are used + * by dt_store_open(). Put them here until better place is found. */ void fid_pack(struct lu_fid_pack *pack, const struct lu_fid *fid, @@ -1473,14 +1664,19 @@ void fid_pack(struct lu_fid_pack *pack, const struct lu_fid *fid, */ CLASSERT(LUSTRE_SEQ_MAX_WIDTH < 0xffffull); - if (fid_is_igif(fid) || - seq > 0xffffffull || oid > 0xffff || fid_ver(fid) != 0) { + /* fid can be packed in six bytes (first byte as length of packed fid, + * three bytes of seq and two bytes of oid). + * this reduces IO overhead specially for OSD Object Index. */ + + if (seq < FID_SEQ_START || + seq > (0xffffffull + FID_SEQ_START) || + oid > 0xffff || fid_ver(fid) != 0) { fid_cpu_to_be(befider, fid); recsize = sizeof *befider; } else { unsigned char *small_befider; - small_befider = (char *)befider; + small_befider = (unsigned char *)befider; small_befider[0] = seq >> 16; small_befider[1] = seq >> 8; @@ -1509,8 +1705,9 @@ int fid_unpack(const struct lu_fid_pack *pack, struct lu_fid *fid) case 6: { const unsigned char *area; - area = pack->fp_area; + area = (unsigned char *)pack->fp_area; fid->f_seq = (area[0] << 16) | (area[1] << 8) | area[2]; + fid->f_seq += FID_SEQ_START; fid->f_oid = (area[3] << 8) | area[4]; fid->f_ver = 0; break; @@ -1522,6 +1719,7 @@ int fid_unpack(const struct lu_fid_pack *pack, struct lu_fid *fid) return result; } EXPORT_SYMBOL(fid_unpack); +#endif /* #ifdef __KERNEL__ */ const char *lu_time_names[LU_TIME_NR] = { [LU_TIME_FIND_LOOKUP] = "find_lookup", @@ -1529,3 +1727,42 @@ const char *lu_time_names[LU_TIME_NR] = { [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; + + 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) { + result = -ENOMEM; + break; + } + } + return result; +} +EXPORT_SYMBOL(lu_kmem_init); + +/** + * Helper function to finalize a number of kmem slab cached at once. Dual to + * lu_kmem_init(). + */ +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); + *caches->ckd_cache = NULL; + } + } +} +EXPORT_SYMBOL(lu_kmem_fini);