X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdd%2Fmdd_orphans.c;h=a1064f88c4c1b6d68e5795bc30a674652f153a12;hp=963493d2dee36fbb77b22c20b790cc42d4afafad;hb=6350af100c204a4db794e7472951ab72c3bc20ef;hpb=e2af7fb3c91dfb13d34d8e1b2f2df8c09621f768 diff --git a/lustre/mdd/mdd_orphans.c b/lustre/mdd/mdd_orphans.c index 963493d..a1064f8 100644 --- a/lustre/mdd/mdd_orphans.c +++ b/lustre/mdd/mdd_orphans.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, 2012, Whamcloud, Inc. + * Copyright (c) 2011, 2013, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -64,7 +64,7 @@ enum { static struct dt_key* orph_key_fill(const struct lu_env *env, const struct lu_fid *lf, __u32 op) { - char *key = mdd_env_info(env)->mti_orph_key; + char *key = mdd_env_info(env)->mti_key; int rc; LASSERT(key); @@ -80,7 +80,7 @@ static struct dt_key* orph_key_fill(const struct lu_env *env, static struct dt_key* orph_key_fill_18(const struct lu_env *env, const struct lu_fid *lf) { - char *key = mdd_env_info(env)->mti_orph_key; + char *key = mdd_env_info(env)->mti_key; int rc; LASSERT(key); @@ -92,29 +92,6 @@ static struct dt_key* orph_key_fill_18(const struct lu_env *env, return ERR_PTR(rc); } -static int orphan_key_to_fid(char *key, struct lu_fid *lf) -{ - int rc = 0; - unsigned int op; - - rc = sscanf(key, ORPHAN_FILE_NAME_FORMAT, - (long long unsigned int *)&lf->f_seq, &lf->f_oid, - &lf->f_ver, &op); - if (rc == 4) - return 0; - - /* build igif */ - rc = sscanf(key, ORPHAN_FILE_NAME_FORMAT_18, - (long long unsigned int *)&lf->f_seq, &lf->f_oid); - if (rc == 2) { - lf->f_ver = 0; - return 0; - } - - CERROR("can not parse orphan file name %s\n", key); - return -EINVAL; -} - static inline void mdd_orphan_write_lock(const struct lu_env *env, struct mdd_device *mdd) { @@ -178,38 +155,41 @@ static inline void mdd_orphan_ref_del(const struct lu_env *env, int orph_declare_index_insert(const struct lu_env *env, - struct mdd_object *obj, - struct thandle *th) + struct mdd_object *obj, + cfs_umode_t mode, struct thandle *th) { - struct mdd_device *mdd = mdo2mdd(&obj->mod_obj); - int rc; + struct mdd_device *mdd = mdo2mdd(&obj->mod_obj); + struct dt_key *key; + int rc; - rc = dt_declare_insert(env, mdd->mdd_orphans, NULL, NULL, th); - if (rc) - return rc; + key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK); - rc = mdo_declare_ref_add(env, obj, th); - if (rc) - return rc; + rc = dt_declare_insert(env, mdd->mdd_orphans, NULL, key, th); + if (rc) + return rc; - if (!S_ISDIR(mdd_object_type(obj))) - return 0; + rc = mdo_declare_ref_add(env, obj, th); + if (rc) + return rc; - rc = mdo_declare_ref_add(env, obj, th); - if (rc) - return rc; + if (!S_ISDIR(mode)) + return 0; - rc = dt_declare_ref_add(env, mdd->mdd_orphans, th); - if (rc) - return rc; + rc = mdo_declare_ref_add(env, obj, th); + if (rc) + return rc; - rc = mdo_declare_index_delete(env, obj, dotdot, th); - if (rc) - return rc; + rc = dt_declare_ref_add(env, mdd->mdd_orphans, th); + if (rc) + return rc; - rc = mdo_declare_index_insert(env, obj, NULL, dotdot, th); + rc = mdo_declare_index_delete(env, obj, dotdot, th); + if (rc) + return rc; - return rc; + rc = mdo_declare_index_insert(env, obj, NULL, dotdot, th); + + return rc; } static int orph_index_insert(const struct lu_env *env, @@ -226,7 +206,6 @@ static int orph_index_insert(const struct lu_env *env, LASSERT(mdd_write_locked(env, obj) != 0); LASSERT(!(obj->mod_flags & ORPHAN_OBJ)); - LASSERT(obj->mod_count > 0); mdd_orphan_write_lock(env, mdd); @@ -263,50 +242,18 @@ out: RETURN(rc); } -/** - * Destroy OSD object on MDD and associated OST objects. - * - * \param obj orphan object - * \param mdd used for sending llog msg to osts - * - * \retval 0 success - * \retval -ve error - */ -static int orphan_object_kill(const struct lu_env *env, - struct mdd_object *obj, - struct mdd_device *mdd, - struct thandle *th) -{ - struct lu_attr *la = &mdd_env_info(env)->mti_la; - int rc = 0; - ENTRY; - - /* No need to lock this object as its recovery phase, and - * no other thread can access it. But we need to lock it - * as its precondition for osd api we using. */ - - mdo_ref_del(env, obj, th); - if (S_ISDIR(mdd_object_type(obj))) { - mdo_ref_del(env, obj, th); - mdd_orphan_ref_del(env, mdd, th); - } else { - /* regular file , cleanup linked ost objects */ - rc = mdd_la_get(env, obj, la, BYPASS_CAPA); - if (rc == 0) - rc = mdd_lov_destroy(env, mdd, obj, la); - } - mdo_destroy(env, obj, th); - RETURN(rc); -} - int orph_declare_index_delete(const struct lu_env *env, struct mdd_object *obj, + struct thandle *th) { struct mdd_device *mdd = mdo2mdd(&obj->mod_obj); + struct dt_key *key; int rc; - rc = dt_declare_delete(env, mdd->mdd_orphans, NULL, th); + key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK); + + rc = dt_declare_delete(env, mdd->mdd_orphans, key, th); if (rc) return rc; @@ -376,28 +323,20 @@ static int orphan_object_destroy(const struct lu_env *env, { struct thandle *th = NULL; struct mdd_device *mdd = mdo2mdd(&obj->mod_obj); - struct md_attr *ma = &mdd_env_info(env)->mti_ma; int rc = 0; ENTRY; - /* init ma */ - ma->ma_lmm_size = mdd_lov_mdsize(env, mdd); - ma->ma_lmm = mdd_max_lmm_get(env, mdd); - ma->ma_cookie_size = mdd_lov_cookiesize(env, mdd); - ma->ma_cookie = mdd_max_cookie_get(env, mdd); - ma->ma_need = MA_INODE | MA_LOV | MA_COOKIE; - ma->ma_valid = 0; - - th = mdd_trans_create(env, mdd); - if (IS_ERR(th)) { - CERROR("Cannot get thandle\n"); - RETURN(-ENOMEM); - } + th = mdd_trans_create(env, mdd); + if (IS_ERR(th)) { + CERROR("Cannot get thandle\n"); + RETURN(PTR_ERR(th)); + } + rc = orph_declare_index_delete(env, obj, th); if (rc) GOTO(stop, rc); - rc = mdd_declare_object_kill(env, obj, ma, th); + rc = mdo_declare_destroy(env, obj, th); if (rc) GOTO(stop, rc); @@ -409,9 +348,14 @@ static int orphan_object_destroy(const struct lu_env *env, if (likely(obj->mod_count == 0)) { mdd_orphan_write_lock(env, mdd); rc = mdd_orphan_delete_obj(env, mdd, key, th); - if (rc == 0) - orphan_object_kill(env, obj, mdd, th); - else + if (rc == 0) { + mdo_ref_del(env, obj, th); + if (S_ISDIR(mdd_object_type(obj))) { + mdo_ref_del(env, obj, th); + mdd_orphan_ref_del(env, mdd, th); + } + rc = mdo_destroy(env, obj, th); + } else CERROR("could not delete object: rc = %d\n",rc); mdd_orphan_write_unlock(env, mdd); } @@ -453,7 +397,7 @@ static int orph_key_test_and_del(const struct lu_env *env, if (rc) /* so replay-single.sh test_37 works */ CERROR("%s: error unlinking orphan "DFID" from " "PENDING: rc = %d\n", - mdd->mdd_obd_dev->obd_name, PFID(lf), rc); + mdd2obd_dev(mdd)->obd_name, PFID(lf), rc); } else { mdd_write_lock(env, mdo, MOR_TGT_CHILD); if (likely(mdo->mod_count > 0)) { @@ -481,15 +425,13 @@ static int orph_key_test_and_del(const struct lu_env *env, * \retval -ve error */ static int orph_index_iterate(const struct lu_env *env, - struct mdd_device *mdd) + struct mdd_device *mdd) { - struct dt_object *dor = mdd->mdd_orphans; - char *mti_key = mdd_env_info(env)->mti_orph_key; - const struct dt_it_ops *iops; - struct dt_it *it; - char *key; - struct lu_fid fid; - int result = 0; + struct dt_object *dor = mdd->mdd_orphans; + struct lu_dirent *ent = &mdd_env_info(env)->mti_ent; + const struct dt_it_ops *iops; + struct dt_it *it; + struct lu_fid fid; int key_sz = 0; int rc; __u64 cookie; @@ -501,7 +443,7 @@ static int orph_index_iterate(const struct lu_env *env, if (IS_ERR(it)) { rc = PTR_ERR(it); CERROR("%s: cannot clean PENDING: rc = %d\n", - mdd->mdd_obd_dev->obd_name, rc); + mdd2obd_dev(mdd)->obd_name, rc); GOTO(out, rc); } @@ -510,57 +452,53 @@ static int orph_index_iterate(const struct lu_env *env, GOTO(out_put, rc); if (rc == 0) { CERROR("%s: error loading iterator to clean PENDING\n", - mdd->mdd_obd_dev->obd_name); + mdd2obd_dev(mdd)->obd_name); /* Index contains no zero key? */ GOTO(out_put, rc = -EIO); } - do { - key = (void *)iops->key(env, it); - if (IS_ERR(key)) { - CERROR("%s: key failed when clean PENDING: rc = %ld\n", - mdd->mdd_obd_dev->obd_name, PTR_ERR(key)); - goto next; - } - key_sz = iops->key_size(env, it); - - /* filter out "." and ".." entries from PENDING dir. */ - if (key_sz < 8) - goto next; - - memcpy(mti_key, key, key_sz); - mti_key[key_sz] = 0; - - if (orphan_key_to_fid(mti_key, &fid)) - goto next; - if (!fid_is_sane(&fid)) { - CERROR("%s: bad FID "DFID" cleaning PENDING\n", - mdd->mdd_obd_dev->obd_name, PFID(&fid)); - goto next; - } - - /* kill orphan object */ - cookie = iops->store(env, it); - iops->put(env, it); - rc = orph_key_test_and_del(env, mdd, &fid, - (struct dt_key *)mti_key); - - /* after index delete reset iterator */ - if (rc == 0) - result = iops->get(env, it, (const void *)""); - else - result = iops->load(env, it, cookie); + do { + key_sz = iops->key_size(env, it); + /* filter out "." and ".." entries from PENDING dir. */ + if (key_sz < 8) + goto next; + + rc = iops->rec(env, it, (struct dt_rec *)ent, LUDA_64BITHASH); + if (rc != 0) { + CERROR("%s: fail to get FID for orphan it: rc = %d\n", + mdd2obd_dev(mdd)->obd_name, rc); + goto next; + } + + fid_le_to_cpu(&fid, &ent->lde_fid); + if (!fid_is_sane(&fid)) { + CERROR("%s: bad FID "DFID" cleaning PENDING\n", + mdd2obd_dev(mdd)->obd_name, PFID(&fid)); + goto next; + } + + /* kill orphan object */ + cookie = iops->store(env, it); + iops->put(env, it); + rc = orph_key_test_and_del(env, mdd, &fid, + (struct dt_key *)ent->lde_name); + + /* after index delete reset iterator */ + if (rc == 0) + rc = iops->get(env, it, (const void *)""); + else + rc = iops->load(env, it, cookie); next: - result = iops->next(env, it); - } while (result == 0); + rc = iops->next(env, it); + } while (rc == 0); - GOTO(out_put, rc = 0); + GOTO(out_put, rc = 0); out_put: - iops->put(env, it); - iops->fini(env, it); + iops->put(env, it); + iops->fini(env, it); out: - return rc; + return rc; } /** @@ -581,26 +519,32 @@ out: */ int orph_index_init(const struct lu_env *env, struct mdd_device *mdd) { - struct lu_fid fid; - struct dt_object *d; - int rc = 0; - ENTRY; - - d = dt_store_open(env, mdd->mdd_child, "", orph_index_name, &fid); - if (!IS_ERR(d)) { - mdd->mdd_orphans = d; - if (!dt_try_as_dir(env, d)) { - rc = -ENOTDIR; - CERROR("\"%s\" is not an index! : rc = %d\n", - orph_index_name, rc); - } - } else { - CERROR("cannot find \"%s\" obj %d\n", - orph_index_name, (int)PTR_ERR(d)); - rc = PTR_ERR(d); - } - - RETURN(rc); + struct lu_fid fid; + struct dt_object *d; + int rc = 0; + + ENTRY; + + /* create PENDING dir */ + fid_zero(&fid); + rc = mdd_local_file_create(env, mdd, &mdd->mdd_local_root_fid, + orph_index_name, S_IFDIR | S_IRUGO | + S_IWUSR | S_IXUGO, &fid); + if (rc < 0) + RETURN(rc); + + d = dt_locate(env, mdd->mdd_child, &fid); + if (IS_ERR(d)) + RETURN(PTR_ERR(d)); + LASSERT(lu_object_exists(&d->do_lu)); + if (!dt_try_as_dir(env, d)) { + CERROR("%s: \"%s\" is not an index: rc = %d\n", + mdd2obd_dev(mdd)->obd_name, orph_index_name, rc); + lu_object_put(env, &d->do_lu); + RETURN(-ENOTDIR); + } + mdd->mdd_orphans = d; + RETURN(0); } void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)