X-Git-Url: https://git.whamcloud.com/?a=blobdiff_plain;f=lustre%2Fptlrpc%2Fnodemap_storage.c;h=72ce072148ab1431d7f06e111216a58694c4b72c;hb=6601661f96325b4971d0d1cb0be0fa01cc2ddc97;hp=b1171202db5d37777abe5eb8043a4cf505948477;hpb=ffe234a077576f0f66f6fe9c165be180133be89a;p=fs%2Flustre-release.git diff --git a/lustre/ptlrpc/nodemap_storage.c b/lustre/ptlrpc/nodemap_storage.c index b117120..72ce072 100644 --- a/lustre/ptlrpc/nodemap_storage.c +++ b/lustre/ptlrpc/nodemap_storage.c @@ -22,7 +22,7 @@ /* * Copyright (C) 2015, Trustees of Indiana University * - * Copyright (c) 2014, Intel Corporation. + * Copyright (c) 2017, Intel Corporation. * * Author: Joshua Walgenbach * Author: Kit Westneat @@ -50,8 +50,8 @@ #include #include #include -#include -#include +#include +#include #include #include #include @@ -71,6 +71,10 @@ static struct nm_config_file *nodemap_mgs_ncf; enum nm_flag_shifts { NM_FL_ALLOW_ROOT_ACCESS = 0x1, NM_FL_TRUST_CLIENT_IDS = 0x2, + NM_FL_DENY_UNKNOWN = 0x4, + NM_FL_MAP_UID_ONLY = 0x8, + NM_FL_MAP_GID_ONLY = 0x10, + NM_FL_ENABLE_AUDIT = 0x20, }; static void nodemap_cluster_key_init(struct nodemap_key *nk, unsigned int nm_id) @@ -85,12 +89,22 @@ static void nodemap_cluster_rec_init(union nodemap_rec *nr, { CLASSERT(sizeof(nr->ncr.ncr_name) == sizeof(nodemap->nm_name)); - strncpy(nr->ncr.ncr_name, nodemap->nm_name, sizeof(nodemap->nm_name)); + strncpy(nr->ncr.ncr_name, nodemap->nm_name, sizeof(nr->ncr.ncr_name)); nr->ncr.ncr_squash_uid = cpu_to_le32(nodemap->nm_squash_uid); nr->ncr.ncr_squash_gid = cpu_to_le32(nodemap->nm_squash_gid); nr->ncr.ncr_flags = cpu_to_le32( - (nodemap->nmf_trust_client_ids ? NM_FL_TRUST_CLIENT_IDS : 0) | - (nodemap->nmf_allow_root_access ? NM_FL_ALLOW_ROOT_ACCESS : 0)); + (nodemap->nmf_trust_client_ids ? + NM_FL_TRUST_CLIENT_IDS : 0) | + (nodemap->nmf_allow_root_access ? + NM_FL_ALLOW_ROOT_ACCESS : 0) | + (nodemap->nmf_deny_unknown ? + NM_FL_DENY_UNKNOWN : 0) | + (nodemap->nmf_map_uid_only ? + NM_FL_MAP_UID_ONLY : 0) | + (nodemap->nmf_map_gid_only ? + NM_FL_MAP_GID_ONLY : 0) | + (nodemap->nmf_enable_audit ? + NM_FL_ENABLE_AUDIT : 0)); } static void nodemap_idmap_key_init(struct nodemap_key *nk, unsigned int nm_id, @@ -148,14 +162,92 @@ static void nodemap_inc_version(const struct lu_env *env, dt_version_set(env, nodemap_idx, ver + 1, th); } +enum ncfc_find_create { + NCFC_CREATE_NEW = 1, +}; + +static struct dt_object *nodemap_cache_find_create(const struct lu_env *env, + struct dt_device *dev, + struct local_oid_storage *los, + enum ncfc_find_create create_new) +{ + struct lu_fid tfid; + struct dt_object *root_obj; + struct dt_object *nm_obj; + int rc = 0; + + rc = dt_root_get(env, dev, &tfid); + if (rc < 0) + GOTO(out, nm_obj = ERR_PTR(rc)); + + root_obj = dt_locate(env, dev, &tfid); + if (unlikely(IS_ERR(root_obj))) + GOTO(out, nm_obj = root_obj); + + rc = dt_lookup_dir(env, root_obj, LUSTRE_NODEMAP_NAME, &tfid); + if (rc == -ENOENT) { + if (dev->dd_rdonly) + GOTO(out_root, nm_obj = ERR_PTR(-EROFS)); + } else if (rc) { + GOTO(out_root, nm_obj = ERR_PTR(rc)); + } else if (dev->dd_rdonly && create_new == NCFC_CREATE_NEW) { + GOTO(out_root, nm_obj = ERR_PTR(-EROFS)); + } + +again: + /* if loading index fails the first time, create new index */ + if (create_new == NCFC_CREATE_NEW && rc != -ENOENT) { + CDEBUG(D_INFO, "removing old index, creating new one\n"); + rc = local_object_unlink(env, dev, root_obj, + LUSTRE_NODEMAP_NAME); + if (rc < 0) { + /* XXX not sure the best way to get obd name. */ + CERROR("cannot destroy nodemap index: rc = %d\n", + rc); + GOTO(out_root, nm_obj = ERR_PTR(rc)); + } + } + + nm_obj = local_index_find_or_create(env, los, root_obj, + LUSTRE_NODEMAP_NAME, + S_IFREG | S_IRUGO | S_IWUSR, + &dt_nodemap_features); + if (IS_ERR(nm_obj)) + GOTO(out_root, nm_obj); + + if (nm_obj->do_index_ops == NULL) { + rc = nm_obj->do_ops->do_index_try(env, nm_obj, + &dt_nodemap_features); + /* even if loading from tgt fails, connecting to MGS will + * rewrite the config + */ + if (rc < 0) { + dt_object_put(env, nm_obj); + + if (create_new == NCFC_CREATE_NEW) + GOTO(out_root, nm_obj = ERR_PTR(rc)); + + CERROR("cannot load nodemap index from disk, creating " + "new index: rc = %d\n", rc); + create_new = NCFC_CREATE_NEW; + goto again; + } + } + +out_root: + dt_object_put(env, root_obj); +out: + return nm_obj; +} + static int nodemap_idx_insert(const struct lu_env *env, struct dt_object *idx, const struct nodemap_key *nk, const union nodemap_rec *nr) { - struct thandle *th; - struct dt_device *dev = lu2dt_dev(idx->do_lu.lo_dev); - int rc; + struct thandle *th; + struct dt_device *dev = lu2dt_dev(idx->do_lu.lo_dev); + int rc; CLASSERT(sizeof(union nodemap_rec) == 32); @@ -181,7 +273,7 @@ static int nodemap_idx_insert(const struct lu_env *env, dt_write_lock(env, idx, 0); rc = dt_insert(env, idx, (const struct dt_rec *)nr, - (const struct dt_key *)nk, th, 1); + (const struct dt_key *)nk, th); nodemap_inc_version(env, idx, th); dt_write_unlock(env, idx); @@ -229,7 +321,7 @@ static int nodemap_idx_update(const struct lu_env *env, GOTO(out_lock, rc); rc = dt_insert(env, idx, (const struct dt_rec *)nr, - (const struct dt_key *)nk, th, 1); + (const struct dt_key *)nk, th); if (rc != 0) GOTO(out_lock, rc); @@ -287,6 +379,7 @@ enum nm_add_update { }; static int nodemap_idx_nodemap_add_update(const struct lu_nodemap *nodemap, + struct dt_object *idx, enum nm_add_update update) { struct nodemap_key nk; @@ -296,11 +389,6 @@ static int nodemap_idx_nodemap_add_update(const struct lu_nodemap *nodemap, ENTRY; - if (nodemap_mgs_ncf == NULL) { - CERROR("cannot add nodemap config to non-existing MGS.\n"); - return -EINVAL; - } - rc = lu_env_init(&env, LCT_LOCAL); if (rc) RETURN(rc); @@ -309,11 +397,9 @@ static int nodemap_idx_nodemap_add_update(const struct lu_nodemap *nodemap, nodemap_cluster_rec_init(&nr, nodemap); if (update == NM_UPDATE) - rc = nodemap_idx_update(&env, nodemap_mgs_ncf->ncf_obj, - &nk, &nr); + rc = nodemap_idx_update(&env, idx, &nk, &nr); else - rc = nodemap_idx_insert(&env, nodemap_mgs_ncf->ncf_obj, - &nk, &nr); + rc = nodemap_idx_insert(&env, idx, &nk, &nr); lu_env_fini(&env); @@ -322,12 +408,24 @@ static int nodemap_idx_nodemap_add_update(const struct lu_nodemap *nodemap, int nodemap_idx_nodemap_add(const struct lu_nodemap *nodemap) { - return nodemap_idx_nodemap_add_update(nodemap, NM_ADD); + if (nodemap_mgs_ncf == NULL) { + CERROR("cannot add nodemap config to non-existing MGS.\n"); + return -EINVAL; + } + + return nodemap_idx_nodemap_add_update(nodemap, nodemap_mgs_ncf->ncf_obj, + NM_ADD); } int nodemap_idx_nodemap_update(const struct lu_nodemap *nodemap) { - return nodemap_idx_nodemap_add_update(nodemap, NM_UPDATE); + if (nodemap_mgs_ncf == NULL) { + CERROR("cannot add nodemap config to non-existing MGS.\n"); + return -EINVAL; + } + + return nodemap_idx_nodemap_add_update(nodemap, nodemap_mgs_ncf->ncf_obj, + NM_UPDATE); } int nodemap_idx_nodemap_del(const struct lu_nodemap *nodemap) @@ -536,6 +634,14 @@ int nodemap_idx_nodemap_activate(bool value) return nodemap_idx_global_add_update(value, NM_UPDATE); } +static enum nodemap_idx_type nodemap_get_key_type(const struct nodemap_key *key) +{ + u32 nodemap_id; + + nodemap_id = le32_to_cpu(key->nk_nodemap_id); + return nm_idx_get_type(nodemap_id); +} + /** * Process a key/rec pair and modify the new configuration. * @@ -554,19 +660,21 @@ static int nodemap_process_keyrec(struct nodemap_config *config, const union nodemap_rec *rec, struct lu_nodemap **recent_nodemap) { - struct lu_nodemap *nodemap = NULL; - enum nodemap_idx_type type; - enum nodemap_id_type id_type; - u8 flags; - u32 nodemap_id; - lnet_nid_t nid[2]; - u32 map[2]; - int rc; + struct lu_nodemap *nodemap = NULL; + enum nodemap_idx_type type; + enum nodemap_id_type id_type; + u8 flags; + u32 nodemap_id; + lnet_nid_t nid[2]; + u32 map[2]; + int rc; + + ENTRY; CLASSERT(sizeof(union nodemap_rec) == 32); nodemap_id = le32_to_cpu(key->nk_nodemap_id); - type = nm_idx_get_type(nodemap_id); + type = nodemap_get_key_type(key); nodemap_id = nm_idx_set_type(nodemap_id, 0); CDEBUG(D_INFO, "found config entry, nm_id %d type %d\n", @@ -605,7 +713,9 @@ static int nodemap_process_keyrec(struct nodemap_config *config, " nodemap_id=%d. nodemap config file corrupt?\n", nodemap_id); break; - case NODEMAP_CLUSTER_IDX: + case NODEMAP_CLUSTER_IDX: { + struct lu_nodemap *old_nm = NULL; + nodemap = cfs_hash_lookup(config->nmc_nodemap_hash, rec->ncr.ncr_name); if (nodemap == NULL) { @@ -640,6 +750,25 @@ static int nodemap_process_keyrec(struct nodemap_config *config, flags & NM_FL_ALLOW_ROOT_ACCESS; nodemap->nmf_trust_client_ids = flags & NM_FL_TRUST_CLIENT_IDS; + nodemap->nmf_deny_unknown = + flags & NM_FL_DENY_UNKNOWN; + nodemap->nmf_map_uid_only = + flags & NM_FL_MAP_UID_ONLY; + nodemap->nmf_map_gid_only = + flags & NM_FL_MAP_GID_ONLY; + nodemap->nmf_enable_audit = + flags & NM_FL_ENABLE_AUDIT; + + /* The fileset should be saved otherwise it will be empty + * every time in case of "NODEMAP_CLUSTER_IDX". */ + mutex_lock(&active_config_lock); + old_nm = nodemap_lookup(rec->ncr.ncr_name); + if (!IS_ERR(old_nm) && old_nm->nm_fileset[0] != '\0') + strlcpy(nodemap->nm_fileset, old_nm->nm_fileset, + sizeof(nodemap->nm_fileset)); + mutex_unlock(&active_config_lock); + if (!IS_ERR(old_nm)) + nodemap_putref(old_nm); if (*recent_nodemap == NULL) { *recent_nodemap = nodemap; @@ -650,6 +779,7 @@ static int nodemap_process_keyrec(struct nodemap_config *config, } nodemap_putref(nodemap); break; + } case NODEMAP_RANGE_IDX: nid[0] = le64_to_cpu(rec->nrr.nrr_start_nid); nid[1] = le64_to_cpu(rec->nrr.nrr_end_nid); @@ -683,21 +813,29 @@ static int nodemap_process_keyrec(struct nodemap_config *config, rc = type; + EXIT; + out: return rc; } +enum nm_config_passes { + NM_READ_CLUSTERS = 0, + NM_READ_ATTRIBUTES = 1, +}; + static int nodemap_load_entries(const struct lu_env *env, struct dt_object *nodemap_idx) { - const struct dt_it_ops *iops; - struct dt_it *it; - struct lu_nodemap *recent_nodemap = NULL; - struct nodemap_config *new_config = NULL; - u64 hash = 0; - bool activate_nodemap = false; - bool loaded_global_idx = false; - int rc = 0; + const struct dt_it_ops *iops; + struct dt_it *it; + struct lu_nodemap *recent_nodemap = NULL; + struct nodemap_config *new_config = NULL; + u64 hash = 0; + bool activate_nodemap = false; + bool loaded_global_idx = false; + enum nm_config_passes cur_pass = NM_READ_CLUSTERS; + int rc = 0; ENTRY; @@ -709,52 +847,83 @@ static int nodemap_load_entries(const struct lu_env *env, GOTO(out, rc = PTR_ERR(it)); rc = iops->load(env, it, hash); + if (rc < 0) + GOTO(out_iops_fini, rc); + + /* rc == 0 means we need to advance to record */ if (rc == 0) { rc = iops->next(env, it); - if (rc != 0) - GOTO(out_iops, rc = 0); + + if (rc < 0) + GOTO(out_iops_put, rc); + /* rc > 0 is eof, will be checked in while below */ + } else { + /* rc == 1, we found initial record and can process below */ + rc = 0; } new_config = nodemap_config_alloc(); if (IS_ERR(new_config)) { rc = PTR_ERR(new_config); new_config = NULL; - GOTO(out_lock, rc); + GOTO(out_iops_put, rc); } - do { + /* rc > 0 is eof, check initial iops->next here as well */ + while (rc == 0) { struct nodemap_key *key; union nodemap_rec rec; + enum nodemap_idx_type key_type; key = (struct nodemap_key *)iops->key(env, it); - rc = iops->rec(env, it, (struct dt_rec *)&rec, 0); - if (rc != -ESTALE) { - if (rc != 0) - GOTO(out_lock, rc); - rc = nodemap_process_keyrec(new_config, key, &rec, - &recent_nodemap); - if (rc < 0) - GOTO(out_lock, rc); - if (rc == NODEMAP_GLOBAL_IDX) - loaded_global_idx = true; + key_type = nodemap_get_key_type((struct nodemap_key *)key); + if ((cur_pass == NM_READ_CLUSTERS && + key_type == NODEMAP_CLUSTER_IDX) || + (cur_pass == NM_READ_ATTRIBUTES && + key_type != NODEMAP_CLUSTER_IDX && + key_type != NODEMAP_EMPTY_IDX)) { + rc = iops->rec(env, it, (struct dt_rec *)&rec, 0); + if (rc != -ESTALE) { + if (rc != 0) + GOTO(out_nodemap_config, rc); + rc = nodemap_process_keyrec(new_config, key, &rec, + &recent_nodemap); + if (rc < 0) + GOTO(out_nodemap_config, rc); + if (rc == NODEMAP_GLOBAL_IDX) + loaded_global_idx = true; + } } do rc = iops->next(env, it); while (rc == -ESTALE); - } while (rc == 0); + + /* move to second pass */ + if (rc > 0 && cur_pass == NM_READ_CLUSTERS) { + cur_pass = NM_READ_ATTRIBUTES; + rc = iops->load(env, it, 0); + if (rc == 0) + rc = iops->next(env, it); + else if (rc > 0) + rc = 0; + else + GOTO(out, rc); + } + } if (rc > 0) rc = 0; -out_lock: +out_nodemap_config: if (rc != 0) nodemap_config_dealloc(new_config); else /* creating new default needs to be done outside dt read lock */ activate_nodemap = true; -out_iops: +out_iops_put: iops->put(env, it); +out_iops_fini: iops->fini(env, it); out: dt_read_unlock(env, nodemap_idx); @@ -775,6 +944,7 @@ out: } else { rc = nodemap_idx_nodemap_add_update( new_config->nmc_default_nodemap, + nodemap_idx, NM_ADD); nodemap_putref(new_config->nmc_default_nodemap); } @@ -798,6 +968,141 @@ out: RETURN(rc); } +/** + * Step through active config and write to disk. + */ +struct dt_object *nodemap_save_config_cache(const struct lu_env *env, + struct dt_device *dev, + struct local_oid_storage *los) +{ + struct dt_object *o; + struct lu_nodemap *nodemap; + struct lu_nodemap *nm_tmp; + struct lu_nid_range *range; + struct lu_nid_range *range_temp; + struct lu_idmap *idmap; + struct lu_idmap *id_tmp; + struct rb_root root; + struct nodemap_key nk; + union nodemap_rec nr; + LIST_HEAD(nodemap_list_head); + int rc = 0, rc2; + + ENTRY; + + /* create a new index file to fill with active config */ + o = nodemap_cache_find_create(env, dev, los, NCFC_CREATE_NEW); + if (IS_ERR(o)) + RETURN(o); + + mutex_lock(&active_config_lock); + + /* convert hash to list so we don't spin */ + cfs_hash_for_each_safe(active_config->nmc_nodemap_hash, + nm_hash_list_cb, &nodemap_list_head); + + list_for_each_entry_safe(nodemap, nm_tmp, &nodemap_list_head, nm_list) { + nodemap_cluster_key_init(&nk, nodemap->nm_id); + nodemap_cluster_rec_init(&nr, nodemap); + + rc2 = nodemap_idx_insert(env, o, &nk, &nr); + if (rc2 < 0) { + rc = rc2; + continue; + } + + down_read(&active_config->nmc_range_tree_lock); + list_for_each_entry_safe(range, range_temp, &nodemap->nm_ranges, + rn_list) { + lnet_nid_t nid[2] = { + range->rn_node.in_extent.start, + range->rn_node.in_extent.end + }; + nodemap_range_key_init(&nk, nodemap->nm_id, + range->rn_id); + nodemap_range_rec_init(&nr, nid); + rc2 = nodemap_idx_insert(env, o, &nk, &nr); + if (rc2 < 0) + rc = rc2; + } + up_read(&active_config->nmc_range_tree_lock); + + /* we don't need to take nm_idmap_lock because active config + * lock prevents changes from happening to nodemaps + */ + root = nodemap->nm_client_to_fs_uidmap; + nm_rbtree_postorder_for_each_entry_safe(idmap, id_tmp, &root, + id_client_to_fs) { + nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_UID, + idmap->id_client); + nodemap_idmap_rec_init(&nr, idmap->id_fs); + rc2 = nodemap_idx_insert(env, o, &nk, &nr); + if (rc2 < 0) + rc = rc2; + } + + root = nodemap->nm_client_to_fs_gidmap; + nm_rbtree_postorder_for_each_entry_safe(idmap, id_tmp, &root, + id_client_to_fs) { + nodemap_idmap_key_init(&nk, nodemap->nm_id, NODEMAP_GID, + idmap->id_client); + nodemap_idmap_rec_init(&nr, idmap->id_fs); + rc2 = nodemap_idx_insert(env, o, &nk, &nr); + if (rc2 < 0) + rc = rc2; + } + } + nodemap_global_key_init(&nk); + nodemap_global_rec_init(&nr, active_config->nmc_nodemap_is_active); + rc2 = nodemap_idx_insert(env, o, &nk, &nr); + if (rc2 < 0) + rc = rc2; + + mutex_unlock(&active_config_lock); + + if (rc < 0) { + dt_object_put(env, o); + o = ERR_PTR(rc); + } + + RETURN(o); +} + +static void nodemap_save_all_caches(void) +{ + struct nm_config_file *ncf; + struct lu_env env; + int rc = 0; + + /* recreating nodemap cache requires fld_thread_key be in env */ + rc = lu_env_init(&env, LCT_MD_THREAD | LCT_DT_THREAD | LCT_MG_THREAD); + if (rc != 0) { + CWARN("cannot init env for nodemap config: rc = %d\n", rc); + return; + } + + mutex_lock(&ncf_list_lock); + list_for_each_entry(ncf, &ncf_list_head, ncf_list) { + struct dt_device *dev = lu2dt_dev(ncf->ncf_obj->do_lu.lo_dev); + struct obd_device *obd = ncf->ncf_obj->do_lu.lo_dev->ld_obd; + struct dt_object *o; + + /* put current config file so save conf can rewrite it */ + dt_object_put_nocache(&env, ncf->ncf_obj); + ncf->ncf_obj = NULL; + + o = nodemap_save_config_cache(&env, dev, ncf->ncf_los); + if (IS_ERR(o)) + CWARN("%s: error writing to nodemap config: rc = %d\n", + obd->obd_name, rc); + else + ncf->ncf_obj = o; + } + mutex_unlock(&ncf_list_lock); + + lu_env_fini(&env); +} + /* tracks if config still needs to be loaded, either from disk or network */ static bool nodemap_config_loaded; static DEFINE_MUTEX(nodemap_config_loaded_lock); @@ -813,6 +1118,7 @@ void nodemap_config_set_active_mgc(struct nodemap_config *config) mutex_lock(&nodemap_config_loaded_lock); nodemap_config_set_active(config); nodemap_config_loaded = true; + nodemap_save_all_caches(); mutex_unlock(&nodemap_config_loaded_lock); } EXPORT_SYMBOL(nodemap_config_set_active_mgc); @@ -829,83 +1135,142 @@ EXPORT_SYMBOL(nodemap_config_set_active_mgc); * \retval -ENOMEM memory allocation failure * \retval -ENOENT error loading nodemap config * \retval -EINVAL error loading nodemap config + * \retval -EEXIST nodemap config already registered for MGS */ -struct nm_config_file *nm_config_file_register(const struct lu_env *env, - struct dt_object *obj, - struct local_oid_storage *los, - enum nm_config_file_type ncf_type) +struct nm_config_file *nm_config_file_register_mgs(const struct lu_env *env, + struct dt_object *obj, + struct local_oid_storage *los) { struct nm_config_file *ncf; int rc = 0; ENTRY; + if (nodemap_mgs_ncf != NULL) + GOTO(out, ncf = ERR_PTR(-EEXIST)); + OBD_ALLOC_PTR(ncf); if (ncf == NULL) - RETURN(ERR_PTR(-ENOMEM)); + GOTO(out, ncf = ERR_PTR(-ENOMEM)); + + /* if loading from cache, prevent activation of MGS config until cache + * loading is done, so disk config is overwritten by MGS config. + */ + mutex_lock(&nodemap_config_loaded_lock); + rc = nodemap_load_entries(env, obj); + if (!rc) + nodemap_config_loaded = true; + mutex_unlock(&nodemap_config_loaded_lock); + + if (rc) { + OBD_FREE_PTR(ncf); + GOTO(out, ncf = ERR_PTR(rc)); + } + + lu_object_get(&obj->do_lu); ncf->ncf_obj = obj; ncf->ncf_los = los; - if (ncf_type == NCFT_MGS) { - nodemap_mgs_ncf = ncf; - } else { - mutex_lock(&ncf_list_lock); - list_add(&ncf->ncf_list, &ncf_list_head); - mutex_unlock(&ncf_list_lock); - } + nodemap_mgs_ncf = ncf; - /* prevent activation of config loaded from MGS until disk is loaded - * so disk config is overwritten by MGS config. - */ +out: + return ncf; +} +EXPORT_SYMBOL(nm_config_file_register_mgs); + +struct nm_config_file *nm_config_file_register_tgt(const struct lu_env *env, + struct dt_device *dev, + struct local_oid_storage *los) +{ + struct nm_config_file *ncf; + struct dt_object *config_obj = NULL; + int rc = 0; + + OBD_ALLOC_PTR(ncf); + if (ncf == NULL) + RETURN(ERR_PTR(-ENOMEM)); + + /* don't load from cache if config already loaded */ mutex_lock(&nodemap_config_loaded_lock); - if (ncf_type == NCFT_MGS || !nodemap_config_loaded) - rc = nodemap_load_entries(env, obj); - nodemap_config_loaded = true; + if (!nodemap_config_loaded) { + config_obj = nodemap_cache_find_create(env, dev, los, 0); + if (IS_ERR(config_obj)) + rc = PTR_ERR(config_obj); + else + rc = nodemap_load_entries(env, config_obj); + + if (!rc) + nodemap_config_loaded = true; + } mutex_unlock(&nodemap_config_loaded_lock); + if (rc) + GOTO(out_ncf, rc); - if (rc < 0) { - if (ncf_type == NCFT_MGS) { - nodemap_mgs_ncf = NULL; - } else { - mutex_lock(&ncf_list_lock); - list_del(&ncf->ncf_list); - mutex_unlock(&ncf_list_lock); - } + /* sync on disk caches w/ loaded config in memory, ncf_obj may change */ + if (!config_obj) { + config_obj = nodemap_save_config_cache(env, dev, los); + if (IS_ERR(config_obj)) + GOTO(out_ncf, rc = PTR_ERR(config_obj)); + } + + ncf->ncf_obj = config_obj; + ncf->ncf_los = los; + + mutex_lock(&ncf_list_lock); + list_add(&ncf->ncf_list, &ncf_list_head); + mutex_unlock(&ncf_list_lock); +out_ncf: + if (rc) { OBD_FREE_PTR(ncf); RETURN(ERR_PTR(rc)); } RETURN(ncf); } -EXPORT_SYMBOL(nm_config_file_register); +EXPORT_SYMBOL(nm_config_file_register_tgt); /** * Deregister a nm_config_file. Should be called by targets during cleanup. * * \param ncf config file to deregister */ -void nm_config_file_deregister(const struct lu_env *env, - struct nm_config_file *ncf, - enum nm_config_file_type ncf_type) +void nm_config_file_deregister_mgs(const struct lu_env *env, + struct nm_config_file *ncf) { ENTRY; + LASSERT(nodemap_mgs_ncf == ncf); + nodemap_mgs_ncf = NULL; if (ncf->ncf_obj) - lu_object_put(env, &ncf->ncf_obj->do_lu); + dt_object_put(env, ncf->ncf_obj); + + OBD_FREE_PTR(ncf); + + EXIT; +} +EXPORT_SYMBOL(nm_config_file_deregister_mgs); + +void nm_config_file_deregister_tgt(const struct lu_env *env, + struct nm_config_file *ncf) +{ + ENTRY; + + if (ncf == NULL) + return; + + mutex_lock(&ncf_list_lock); + list_del(&ncf->ncf_list); + mutex_unlock(&ncf_list_lock); + + if (ncf->ncf_obj) + dt_object_put(env, ncf->ncf_obj); - if (ncf_type == NCFT_TGT) { - mutex_lock(&ncf_list_lock); - list_del(&ncf->ncf_list); - mutex_unlock(&ncf_list_lock); - } else { - nodemap_mgs_ncf = NULL; - } OBD_FREE_PTR(ncf); EXIT; } -EXPORT_SYMBOL(nm_config_file_deregister); +EXPORT_SYMBOL(nm_config_file_deregister_tgt); int nodemap_process_idx_pages(struct nodemap_config *config, union lu_page *lip, struct lu_nodemap **recent_nodemap) @@ -945,6 +1310,109 @@ int nodemap_process_idx_pages(struct nodemap_config *config, union lu_page *lip, } EXPORT_SYMBOL(nodemap_process_idx_pages); +static int nodemap_page_build(const struct lu_env *env, union lu_page *lp, + size_t nob, const struct dt_it_ops *iops, + struct dt_it *it, __u32 attr, void *arg) +{ + struct idx_info *ii = (struct idx_info *)arg; + struct lu_idxpage *lip = &lp->lp_idx; + char *entry; + size_t size = ii->ii_keysize + ii->ii_recsize; + int rc; + ENTRY; + + if (nob < LIP_HDR_SIZE) + return -EINVAL; + + /* initialize the header of the new container */ + memset(lip, 0, LIP_HDR_SIZE); + lip->lip_magic = LIP_MAGIC; + nob -= LIP_HDR_SIZE; + + entry = lip->lip_entries; + do { + char *tmp_entry = entry; + struct dt_key *key; + __u64 hash; + enum nodemap_idx_type key_type; + + /* fetch 64-bit hash value */ + hash = iops->store(env, it); + ii->ii_hash_end = hash; + + if (OBD_FAIL_CHECK(OBD_FAIL_OBD_IDX_READ_BREAK)) { + if (lip->lip_nr != 0) + GOTO(out, rc = 0); + } + + if (nob < size) { + if (lip->lip_nr == 0) + GOTO(out, rc = -EINVAL); + GOTO(out, rc = 0); + } + + key = iops->key(env, it); + key_type = nodemap_get_key_type((struct nodemap_key *)key); + + /* on the first pass, get only the cluster types. On second + * pass, get all the rest */ + if ((ii->ii_attrs == NM_READ_CLUSTERS && + key_type == NODEMAP_CLUSTER_IDX) || + (ii->ii_attrs == NM_READ_ATTRIBUTES && + key_type != NODEMAP_CLUSTER_IDX && + key_type != NODEMAP_EMPTY_IDX)) { + memcpy(tmp_entry, key, ii->ii_keysize); + tmp_entry += ii->ii_keysize; + + /* and finally the record */ + rc = iops->rec(env, it, (struct dt_rec *)tmp_entry, + attr); + if (rc != -ESTALE) { + if (rc != 0) + GOTO(out, rc); + + /* hash/key/record successfully copied! */ + lip->lip_nr++; + if (unlikely(lip->lip_nr == 1 && + ii->ii_count == 0)) + ii->ii_hash_start = hash; + + entry = tmp_entry + ii->ii_recsize; + nob -= size; + } + } + + /* move on to the next record */ + do { + rc = iops->next(env, it); + } while (rc == -ESTALE); + + /* move to second pass */ + if (rc > 0 && ii->ii_attrs == NM_READ_CLUSTERS) { + ii->ii_attrs = NM_READ_ATTRIBUTES; + rc = iops->load(env, it, 0); + if (rc == 0) + rc = iops->next(env, it); + else if (rc > 0) + rc = 0; + else + GOTO(out, rc); + } + + } while (rc == 0); + + GOTO(out, rc); +out: + if (rc >= 0 && lip->lip_nr > 0) + /* one more container */ + ii->ii_count++; + if (rc > 0) + /* no more entries */ + ii->ii_hash_end = II_END_OFF; + return rc; +} + + int nodemap_index_read(struct lu_env *env, struct nm_config_file *ncf, struct idx_info *ii, @@ -960,13 +1428,13 @@ int nodemap_index_read(struct lu_env *env, dt_read_lock(env, nodemap_idx, 0); version = dt_version_get(env, nodemap_idx); if (rdpg->rp_hash != 0 && ii->ii_version != version) { - CDEBUG(D_INFO, "nodemap config changed while sending, " - "old "LPU64", new "LPU64"\n", + CDEBUG(D_INFO, "nodemap config changed inflight, old %llu, new %llu\n", ii->ii_version, version); ii->ii_hash_end = 0; } else { - rc = dt_index_walk(env, nodemap_idx, rdpg, NULL, ii); + rc = dt_index_walk(env, nodemap_idx, rdpg, nodemap_page_build, + ii); CDEBUG(D_INFO, "walked index, hashend %llx\n", ii->ii_hash_end); } @@ -1012,8 +1480,8 @@ int nodemap_get_config_req(struct obd_device *mgs_obd, RETURN(-EINVAL); rdpg.rp_count = (body->mcb_units << body->mcb_bits); - rdpg.rp_npages = (rdpg.rp_count + PAGE_CACHE_SIZE - 1) >> - PAGE_CACHE_SHIFT; + rdpg.rp_npages = (rdpg.rp_count + PAGE_SIZE - 1) >> + PAGE_SHIFT; if (rdpg.rp_npages > PTLRPC_MAX_BRW_PAGES) RETURN(-EINVAL); @@ -1025,7 +1493,7 @@ int nodemap_get_config_req(struct obd_device *mgs_obd, if (rdpg.rp_pages == NULL) RETURN(-ENOMEM); for (i = 0; i < rdpg.rp_npages; i++) { - rdpg.rp_pages[i] = alloc_page(GFP_IOFS); + rdpg.rp_pages[i] = alloc_page(GFP_NOFS); if (rdpg.rp_pages[i] == NULL) GOTO(out, rc = -ENOMEM); } @@ -1034,6 +1502,7 @@ int nodemap_get_config_req(struct obd_device *mgs_obd, nodemap_ii.ii_magic = IDX_INFO_MAGIC; nodemap_ii.ii_flags = II_FL_NOHASH; nodemap_ii.ii_version = rqexp_ted->ted_nodemap_version; + nodemap_ii.ii_attrs = body->mcb_nm_cur_pass; bytes = nodemap_index_read(req->rq_svc_thread->t_env, mgs_obd->u.obt.obt_nodemap_config_file, @@ -1047,9 +1516,9 @@ int nodemap_get_config_req(struct obd_device *mgs_obd, if (res == NULL) GOTO(out, rc = -EINVAL); res->mcr_offset = nodemap_ii.ii_hash_end; - res->mcr_size = bytes; + res->mcr_nm_cur_pass = nodemap_ii.ii_attrs; - page_count = (bytes + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT; + page_count = (bytes + PAGE_SIZE - 1) >> PAGE_SHIFT; LASSERT(page_count <= rdpg.rp_count); desc = ptlrpc_prep_bulk_exp(req, page_count, 1, PTLRPC_BULK_PUT_SOURCE | @@ -1061,8 +1530,8 @@ int nodemap_get_config_req(struct obd_device *mgs_obd, for (i = 0; i < page_count && bytes > 0; i++) { ptlrpc_prep_bulk_page_pin(desc, rdpg.rp_pages[i], 0, - min_t(int, bytes, PAGE_CACHE_SIZE)); - bytes -= PAGE_CACHE_SIZE; + min_t(int, bytes, PAGE_SIZE)); + bytes -= PAGE_SIZE; } rc = target_bulk_io(req->rq_export, desc, &lwi);