X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Ffld%2Ffld_index.c;h=4226438b43d0cc2d47fd4f74922dcf79279d6d26;hp=0dd72cf9f92e2e7d1d8ed254cf2104250c04dfe7;hb=1ecf5bd5a3bece3d615d037a332792e360b49a09;hpb=08aa217ce49aba1ded52e0f7adb8a607035123fd diff --git a/lustre/fld/fld_index.c b/lustre/fld/fld_index.c index 0dd72cf..4226438 100644 --- a/lustre/fld/fld_index.c +++ b/lustre/fld/fld_index.c @@ -27,7 +27,7 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2011, 2013, Intel Corporation. + * Copyright (c) 2011, 2014, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -41,28 +41,15 @@ #define DEBUG_SUBSYSTEM S_FLD -#ifdef __KERNEL__ -# include -# include -# include -#else /* __KERNEL__ */ -# include -#endif - -#include -#include -#include +#include +#include #include -#include - #include -#include -#include #include #include #include "fld_internal.h" -const char fld_index_name[] = "fld"; +static const char fld_index_name[] = "fld"; static const struct lu_seq_range IGIF_FLD_RANGE = { .lsr_start = FID_SEQ_IGIF, @@ -71,10 +58,24 @@ static const struct lu_seq_range IGIF_FLD_RANGE = { .lsr_flags = LU_SEQ_RANGE_MDT }; -const struct dt_index_features fld_index_features = { +static const struct lu_seq_range DOT_LUSTRE_FLD_RANGE = { + .lsr_start = FID_SEQ_DOT_LUSTRE, + .lsr_end = FID_SEQ_DOT_LUSTRE + 1, + .lsr_index = 0, + .lsr_flags = LU_SEQ_RANGE_MDT +}; + +static const struct lu_seq_range ROOT_FLD_RANGE = { + .lsr_start = FID_SEQ_ROOT, + .lsr_end = FID_SEQ_ROOT + 1, + .lsr_index = 0, + .lsr_flags = LU_SEQ_RANGE_MDT +}; + +static const struct dt_index_features fld_index_features = { .dif_flags = DT_IND_UPDATE, - .dif_keysize_min = sizeof(seqno_t), - .dif_keysize_max = sizeof(seqno_t), + .dif_keysize_min = sizeof(u64), + .dif_keysize_max = sizeof(u64), .dif_recsize_min = sizeof(struct lu_seq_range), .dif_recsize_max = sizeof(struct lu_seq_range), .dif_ptrsize = 4 @@ -102,7 +103,7 @@ int fld_declare_index_create(const struct lu_env *env, rc = fld_index_lookup(env, fld, new_range->lsr_start, range); if (rc == 0) { /* In case of duplicate entry, the location must be same */ - LASSERT((range_compare_loc(new_range, range) == 0)); + LASSERT((lu_seq_range_compare_loc(new_range, range) == 0)); GOTO(out, rc = -EEXIST); } @@ -115,7 +116,7 @@ int fld_declare_index_create(const struct lu_env *env, /* Check for merge case, since the fld entry can only be increamental, * so we will only check whether it can be merged from the left. */ if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 && - range_compare_loc(new_range, range) == 0) { + lu_seq_range_compare_loc(new_range, range) == 0) { range_cpu_to_be(tmp, range); rc = dt_declare_delete(env, fld->lsf_obj, (struct dt_key *)&tmp->lsr_start, th); @@ -124,10 +125,10 @@ int fld_declare_index_create(const struct lu_env *env, fld->lsf_name, PRANGE(range), rc); GOTO(out, rc); } - memcpy(tmp, new_range, sizeof(*new_range)); + *tmp = *new_range; tmp->lsr_start = range->lsr_start; } else { - memcpy(tmp, new_range, sizeof(*new_range)); + *tmp = *new_range; } range_cpu_to_be(tmp, tmp); @@ -153,6 +154,8 @@ out: * changed between declare and create. * Because the fld entry can only be increamental, so we will only check * whether it can be merged from the left. + * + * Caller must hold fld->lsf_lock **/ int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld, const struct lu_seq_range *new_range, struct thandle *th) @@ -167,7 +170,7 @@ int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld, info = lu_context_key_get(&env->le_ctx, &fld_thread_key); - LASSERT_MUTEX_LOCKED(&fld->lsf_lock); + LASSERT(mutex_is_locked(&fld->lsf_lock)); range = &info->fti_lrange; memset(range, 0, sizeof(*range)); @@ -179,23 +182,22 @@ int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld, } if (new_range->lsr_start == range->lsr_end && range->lsr_end != 0 && - range_compare_loc(new_range, range) == 0) { + lu_seq_range_compare_loc(new_range, range) == 0) { range_cpu_to_be(tmp, range); rc = dt_delete(env, fld->lsf_obj, - (struct dt_key *)&tmp->lsr_start, th, - BYPASS_CAPA); + (struct dt_key *)&tmp->lsr_start, th); if (rc != 0) GOTO(out, rc); - memcpy(tmp, new_range, sizeof(*new_range)); + *tmp = *new_range; tmp->lsr_start = range->lsr_start; deleted = 1; } else { - memcpy(tmp, new_range, sizeof(*new_range)); + *tmp = *new_range; } range_cpu_to_be(tmp, tmp); rc = dt_insert(env, fld->lsf_obj, (struct dt_rec *)tmp, - (struct dt_key *)&tmp->lsr_start, th, BYPASS_CAPA, 1); + (struct dt_key *)&tmp->lsr_start, th, 1); if (rc != 0) { CERROR("%s: insert range "DRANGE" failed: rc = %d\n", fld->lsf_name, PRANGE(new_range), rc); @@ -206,11 +208,11 @@ int fld_index_create(const struct lu_env *env, struct lu_server_fld *fld, if (IS_ERR(flde)) GOTO(out, rc = PTR_ERR(flde)); - spin_lock(&fld->lsf_cache->fci_lock); + write_lock(&fld->lsf_cache->fci_lock); if (deleted) fld_cache_delete_nolock(fld->lsf_cache, new_range); rc = fld_cache_insert_nolock(fld->lsf_cache, flde); - spin_unlock(&fld->lsf_cache->fci_lock); + write_unlock(&fld->lsf_cache->fci_lock); if (rc) OBD_FREE_PTR(flde); out: @@ -229,7 +231,7 @@ out: * \retval -ve other error; */ int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld, - seqno_t seq, struct lu_seq_range *range) + u64 seq, struct lu_seq_range *range) { struct lu_seq_range *fld_rec; struct fld_thread_info *info; @@ -242,11 +244,11 @@ int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld, rc = fld_cache_lookup(fld->lsf_cache, seq, fld_rec); if (rc == 0) { - *range = *fld_rec; - if (range_within(range, seq)) - rc = 0; - else - rc = -ENOENT; + *range = *fld_rec; + if (lu_seq_range_within(range, seq)) + rc = 0; + else + rc = -ENOENT; } CDEBUG(D_INFO, "%s: lookup seq = "LPX64" range : "DRANGE" rc = %d\n", @@ -255,18 +257,34 @@ int fld_index_lookup(const struct lu_env *env, struct lu_server_fld *fld, RETURN(rc); } -static int fld_insert_igif_fld(struct lu_server_fld *fld, - const struct lu_env *env) +/** + * insert entry in fld store. + * + * \param env relevant lu_env + * \param fld fld store + * \param range range to be inserted + * + * \retval 0 success + * \retval -ve error + * + * Caller must hold fld->lsf_lock + **/ + +int fld_insert_entry(const struct lu_env *env, + struct lu_server_fld *fld, + const struct lu_seq_range *range) { struct thandle *th; int rc; ENTRY; + LASSERT(mutex_is_locked(&fld->lsf_lock)); + th = dt_trans_create(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev)); if (IS_ERR(th)) RETURN(PTR_ERR(th)); - rc = fld_declare_index_create(env, fld, &IGIF_FLD_RANGE, th); + rc = fld_declare_index_create(env, fld, range, th); if (rc != 0) { if (rc == -EEXIST) rc = 0; @@ -278,16 +296,35 @@ static int fld_insert_igif_fld(struct lu_server_fld *fld, if (rc) GOTO(out, rc); - rc = fld_index_create(env, fld, &IGIF_FLD_RANGE, th); + rc = fld_index_create(env, fld, range, th); if (rc == -EEXIST) rc = 0; out: dt_trans_stop(env, lu2dt_dev(fld->lsf_obj->do_lu.lo_dev), th); RETURN(rc); } +EXPORT_SYMBOL(fld_insert_entry); + +static int fld_insert_special_entries(const struct lu_env *env, + struct lu_server_fld *fld) +{ + int rc; + + rc = fld_insert_entry(env, fld, &IGIF_FLD_RANGE); + if (rc != 0) + RETURN(rc); + + rc = fld_insert_entry(env, fld, &DOT_LUSTRE_FLD_RANGE); + if (rc != 0) + RETURN(rc); + + rc = fld_insert_entry(env, fld, &ROOT_FLD_RANGE); + + RETURN(rc); +} int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld, - struct dt_device *dt) + struct dt_device *dt, int type) { struct dt_object *dt_obj = NULL; struct lu_fid fid; @@ -298,6 +335,7 @@ int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld, struct dt_it *it; const struct dt_it_ops *iops; int rc; + __u32 index; ENTRY; info = lu_context_key_get(&env->le_ctx, &fld_thread_key); @@ -308,33 +346,36 @@ int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld, if (attr == NULL) RETURN(-ENOMEM); - memset(attr, 0, sizeof(attr)); + memset(attr, 0, sizeof(*attr)); attr->la_valid = LA_MODE; attr->la_mode = S_IFREG | 0666; dof.dof_type = DFT_INDEX; dof.u.dof_idx.di_feat = &fld_index_features; - dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr); + dt_obj = dt_locate(env, dt, &fid); if (IS_ERR(dt_obj)) { rc = PTR_ERR(dt_obj); - CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name, - fld_index_name, rc); dt_obj = NULL; GOTO(out, rc); } - fld->lsf_obj = dt_obj; - rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features); - if (rc == 0) { - LASSERT(dt_obj->do_index_ops != NULL); - mutex_lock(&fld->lsf_lock); - rc = fld_insert_igif_fld(fld, env); - mutex_unlock(&fld->lsf_lock); - if (rc != 0) { - CERROR("insert igif in fld! = %d\n", rc); + LASSERT(dt_obj != NULL); + if (!dt_object_exists(dt_obj)) { + lu_object_put(env, &dt_obj->do_lu); + dt_obj = dt_find_or_create(env, dt, &fid, &dof, attr); + fld->lsf_new = 1; + if (IS_ERR(dt_obj)) { + rc = PTR_ERR(dt_obj); + CERROR("%s: Can't find \"%s\" obj %d\n", fld->lsf_name, + fld_index_name, rc); + dt_obj = NULL; GOTO(out, rc); } - } else { + } + + fld->lsf_obj = dt_obj; + rc = dt_obj->do_ops->do_index_try(env, dt_obj, &fld_index_features); + if (rc != 0) { CERROR("%s: File \"%s\" is not an index: rc = %d!\n", fld->lsf_name, fld_index_name, rc); GOTO(out, rc); @@ -343,7 +384,7 @@ int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld, range = &info->fti_rec; /* Load fld entry to cache */ iops = &dt_obj->do_index_ops->dio_it; - it = iops->init(env, dt_obj, 0, NULL); + it = iops->init(env, dt_obj, 0); if (IS_ERR(it)) GOTO(out, rc = PTR_ERR(it)); @@ -351,28 +392,50 @@ int fld_index_init(const struct lu_env *env, struct lu_server_fld *fld, if (rc < 0) GOTO(out_it_fini, rc); - do { - rc = iops->rec(env, it, (struct dt_rec *)range, 0); - if (rc != 0) - GOTO(out_it_fini, rc); - - LASSERT(range != NULL); - range_be_to_cpu(range, range); - rc = fld_cache_insert(fld->lsf_cache, range); - if (rc != 0) - GOTO(out_it_fini, rc); - rc = iops->next(env, it); + if (rc > 0) { + /* Load FLD entry into server cache */ + do { + rc = iops->rec(env, it, (struct dt_rec *)range, 0); + if (rc != 0) + GOTO(out_it_put, rc); + LASSERT(range != NULL); + range_be_to_cpu(range, range); + rc = fld_cache_insert(fld->lsf_cache, range); + if (rc != 0) + GOTO(out_it_put, rc); + rc = iops->next(env, it); + } while (rc == 0); + } else { + fld->lsf_new = 1; + } - } while (rc == 0); - rc = 0; + rc = fld_name_to_index(fld->lsf_name, &index); + if (rc < 0) + GOTO(out_it_put, rc); + else + rc = 0; + if (index == 0 && type == LU_SEQ_RANGE_MDT) { + /* Note: fld_insert_entry will detect whether these + * special entries already exist inside FLDB */ + mutex_lock(&fld->lsf_lock); + rc = fld_insert_special_entries(env, fld); + mutex_unlock(&fld->lsf_lock); + if (rc != 0) { + CERROR("%s: insert special entries failed!: rc = %d\n", + fld->lsf_name, rc); + GOTO(out_it_put, rc); + } + } +out_it_put: + iops->put(env, it); out_it_fini: iops->fini(env, it); out: if (attr != NULL) OBD_FREE_PTR(attr); - if (rc != 0) { + if (rc < 0) { if (dt_obj != NULL) lu_object_put(env, &dt_obj->do_lu); fld->lsf_obj = NULL; @@ -390,3 +453,60 @@ void fld_index_fini(const struct lu_env *env, struct lu_server_fld *fld) } EXIT; } + +int fld_server_read(const struct lu_env *env, struct lu_server_fld *fld, + struct lu_seq_range *range, void *data, int data_len) +{ + struct lu_seq_range_array *lsra = data; + struct fld_thread_info *info; + struct dt_object *dt_obj = fld->lsf_obj; + struct lu_seq_range *entry; + struct dt_it *it; + const struct dt_it_ops *iops; + int rc; + + ENTRY; + + lsra->lsra_count = 0; + iops = &dt_obj->do_index_ops->dio_it; + it = iops->init(env, dt_obj, 0); + if (IS_ERR(it)) + RETURN(PTR_ERR(it)); + + rc = iops->load(env, it, range->lsr_end); + if (rc <= 0) + GOTO(out_it_fini, rc); + + info = lu_context_key_get(&env->le_ctx, &fld_thread_key); + LASSERT(info != NULL); + entry = &info->fti_rec; + do { + rc = iops->rec(env, it, (struct dt_rec *)entry, 0); + if (rc != 0) + GOTO(out_it_put, rc); + + if (offsetof(typeof(*lsra), lsra_lsr[lsra->lsra_count + 1]) > + data_len) + GOTO(out, rc = -EAGAIN); + + range_be_to_cpu(entry, entry); + if (entry->lsr_index == range->lsr_index && + entry->lsr_flags == range->lsr_flags && + entry->lsr_start > range->lsr_start) { + lsra->lsra_lsr[lsra->lsra_count] = *entry; + lsra->lsra_count++; + } + + rc = iops->next(env, it); + } while (rc == 0); + if (rc > 0) + rc = 0; +out: + range_array_cpu_to_le(lsra, lsra); +out_it_put: + iops->put(env, it); +out_it_fini: + iops->fini(env, it); + + RETURN(rc); +}