X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdd%2Fmdd_orphans.c;h=3fa6b8bcc4dc111b242bf017303c5c4720de8e09;hp=8ab42cac5f0dd893866c6645adaf6012003a1ba5;hb=e2ac6e1eaa108eef3493837e9bd881629582ea1d;hpb=5c4573e327c5d027a6b877f75910767244161a1f

diff --git a/lustre/mdd/mdd_orphans.c b/lustre/mdd/mdd_orphans.c
index 8ab42ca..3fa6b8bc 100644
--- a/lustre/mdd/mdd_orphans.c
+++ b/lustre/mdd/mdd_orphans.c
@@ -15,11 +15,7 @@
  *
  * You should have received a copy of the GNU General Public License
  * version 2 along with this program; If not, see
- * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
- *
- * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
- * CA 95054 USA or visit www.sun.com if you need additional information or
- * have any questions.
+ * http://www.gnu.org/licenses/gpl-2.0.html
  *
  * GPL HEADER END
  */
@@ -27,7 +23,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, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -45,127 +41,77 @@
 
 #include <obd.h>
 #include <obd_class.h>
-#include <lustre_ver.h>
 #include <obd_support.h>
 #include <lustre_fid.h>
 #include "mdd_internal.h"
 
-const char orph_index_name[] = "PENDING";
+static const char mdd_orphan_index_name[] = MDT_ORPHAN_DIR;
 static const char dotdot[] = "..";
 
 enum {
-        ORPH_OP_UNLINK,
-        ORPH_OP_TRUNCATE
+	ORPH_OP_UNLINK,
 };
 
-#define ORPHAN_FILE_NAME_FORMAT         "%016llx:%08x:%08x:%2x"
-#define ORPHAN_FILE_NAME_FORMAT_18      "%llx:%08x"
-
-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_key;
-        int rc;
-
-        LASSERT(key);
-        rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT,
-                      (long long unsigned int)fid_seq(lf),
-                      fid_oid(lf), fid_ver(lf), op);
-        if (rc > 0)
-                return (struct dt_key*) key;
-        else
-                return ERR_PTR(rc);
-}
+/* obsolete after 2.11, needed for upgrades from older 2.x versions */
+#define ORPHAN_FILE_NAME_FORMAT_20      "%016llx:%08x:%08x:%2x"
 
-static struct dt_key* orph_key_fill_18(const struct lu_env *env,
-                                       const struct lu_fid *lf)
+static struct dt_key *mdd_orphan_key_fill(const struct lu_env *env,
+					  const struct lu_fid *lf)
 {
 	char *key = mdd_env_info(env)->mti_key;
-        int rc;
-
-        LASSERT(key);
-        rc = snprintf(key, NAME_MAX + 1, ORPHAN_FILE_NAME_FORMAT_18,
-                      (unsigned long long)fid_seq(lf), fid_oid(lf));
-        if (rc > 0)
-                return (struct dt_key*) key;
-        else
-                return ERR_PTR(rc);
-}
 
-static inline void mdd_orphan_write_lock(const struct lu_env *env,
-                                    struct mdd_device *mdd)
-{
+	LASSERT(key);
+	snprintf(key, sizeof(mdd_env_info(env)->mti_key),
+		 DFID_NOBRACE, PFID(lf));
 
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        dor->do_ops->do_write_lock(env, dor, MOR_TGT_ORPHAN);
+	return (struct dt_key *)key;
 }
 
-static inline void mdd_orphan_write_unlock(const struct lu_env *env,
-                                           struct mdd_device *mdd)
+/* compatibility with orphan files created in versions before 2.11 */
+static struct dt_key *mdd_orphan_key_fill_20(const struct lu_env *env,
+					     const struct lu_fid *lf)
 {
+	char *key = mdd_env_info(env)->mti_key;
 
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        dor->do_ops->do_write_unlock(env, dor);
-}
+	LASSERT(key);
+	snprintf(key, sizeof(mdd_env_info(env)->mti_key),
+		 ORPHAN_FILE_NAME_FORMAT_20,
+		 fid_seq(lf), fid_oid(lf), fid_ver(lf), ORPH_OP_UNLINK);
 
-static inline int mdd_orphan_insert_obj(const struct lu_env *env,
-                                        struct mdd_device *mdd,
-                                        struct mdd_object *obj,
-                                        __u32 op,
-                                        struct thandle *th)
-{
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        const struct lu_fid     *lf     = mdo2fid(obj);
-        struct dt_key           *key    = orph_key_fill(env, lf, op);
-        ENTRY;
-
-        return  dor->do_index_ops->dio_insert(env, dor,
-                                              (struct dt_rec *)lf,
-                                              key, th,
-                                              BYPASS_CAPA, 1);
+	return (struct dt_key *)key;
 }
 
-static inline int mdd_orphan_delete_obj(const struct lu_env *env,
-                                        struct mdd_device  *mdd ,
-                                        struct dt_key *key,
-                                        struct thandle *th)
+static inline int mdd_orphan_insert_obj(const struct lu_env *env,
+					struct mdd_device *mdd,
+					struct mdd_object *obj,
+					struct thandle *th)
 {
-        struct dt_object        *dor    = mdd->mdd_orphans;
+	struct dt_insert_rec *rec = &mdd_env_info(env)->mti_dt_rec;
+	struct dt_object *dor = mdd->mdd_orphans;
+	const struct lu_fid *lf = mdo2fid(obj);
+	struct dt_key *key = mdd_orphan_key_fill(env, lf);
 
-        return  dor->do_index_ops->dio_delete(env, dor,
-                                              key, th,
-                                              BYPASS_CAPA);
-}
+	rec->rec_fid = lf;
+	rec->rec_type = mdd_object_type(obj);
 
-static inline void mdd_orphan_ref_add(const struct lu_env *env,
-                                 struct mdd_device *mdd,
-                                 struct thandle *th)
-{
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        dor->do_ops->do_ref_add(env, dor, th);
+	return dt_insert(env, dor, (const struct dt_rec *)rec, key, th);
 }
 
-static inline void mdd_orphan_ref_del(const struct lu_env *env,
-                                 struct mdd_device *mdd,
-                                 struct thandle *th)
-{
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        dor->do_ops->do_ref_del(env, dor, th);
-}
-
-
-int orph_declare_index_insert(const struct lu_env *env,
-			      struct mdd_object *obj,
+int mdd_orphan_declare_insert(const struct lu_env *env, struct mdd_object *obj,
 			      umode_t mode, struct thandle *th)
 {
-	struct mdd_device	*mdd = mdo2mdd(&obj->mod_obj);
-	struct dt_key		*key;
-	int			rc;
-
-	key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK);
-
-	rc = dt_declare_insert(env, mdd->mdd_orphans, NULL, key, th);
-	if (rc)
+	struct dt_insert_rec *rec = &mdd_env_info(env)->mti_dt_rec;
+	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+	struct dt_key *key;
+	int rc;
+
+	key = mdd_orphan_key_fill(env, mdo2fid(obj));
+
+	rec->rec_fid = mdo2fid(obj);
+	rec->rec_type = mode;
+	rc = dt_declare_insert(env, mdd->mdd_orphans,
+			       (const struct dt_rec *)rec, key, th);
+	if (rc != 0)
 		return rc;
 
 	rc = mdo_declare_ref_add(env, obj, th);
@@ -187,75 +133,85 @@ int orph_declare_index_insert(const struct lu_env *env,
 	if (rc)
 		return rc;
 
-	rc = mdo_declare_index_insert(env, obj, NULL, dotdot, th);
+	rc = mdo_declare_index_insert(env, obj,
+				      lu_object_fid(&mdd->mdd_orphans->do_lu),
+				      S_IFDIR, dotdot, th);
 
 	return rc;
 }
 
-static int orph_index_insert(const struct lu_env *env,
-                             struct mdd_object *obj,
-                             __u32 op,
-                             struct thandle *th)
+/**
+ *  add an orphan \a obj to the orphan index.
+ *  \param obj file or directory.
+ *  \param th  transaction for index insert.
+ *
+ *  \pre obj nlink == 0 && obj->mod_count != 0
+ *
+ *  \retval 0  success
+ *  \retval  -ve index operation error.
+ */
+int mdd_orphan_insert(const struct lu_env *env, struct mdd_object *obj,
+		      struct thandle *th)
 {
-        struct mdd_device       *mdd    = mdo2mdd(&obj->mod_obj);
-        struct dt_object        *dor    = mdd->mdd_orphans;
-        const struct lu_fid     *lf_dor = lu_object_fid(&dor->do_lu);
-        struct dt_object        *next   = mdd_object_child(obj);
-        int rc;
-        ENTRY;
+	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+	struct dt_object *dor = mdd->mdd_orphans;
+	const struct lu_fid *lf_dor = lu_object_fid(&dor->do_lu);
+	struct dt_object *next = mdd_object_child(obj);
+	struct dt_insert_rec *rec = &mdd_env_info(env)->mti_dt_rec;
+	int rc;
+	ENTRY;
 
-        LASSERT(mdd_write_locked(env, obj) != 0);
-        LASSERT(!(obj->mod_flags & ORPHAN_OBJ));
+	LASSERT(mdd_write_locked(env, obj) != 0);
+	LASSERT(!(obj->mod_flags & ORPHAN_OBJ));
 
-        mdd_orphan_write_lock(env, mdd);
+	dt_write_lock(env, mdd->mdd_orphans, DT_TGT_ORPHAN);
 
-        rc = mdd_orphan_insert_obj(env, mdd, obj, op, th);
-        if (rc)
-                GOTO(out, rc);
+	rc = mdd_orphan_insert_obj(env, mdd, obj, th);
+	if (rc)
+		GOTO(out, rc);
 
-        mdo_ref_add(env, obj, th);
-        if (!S_ISDIR(mdd_object_type(obj)))
+	mdo_ref_add(env, obj, th);
+	if (!S_ISDIR(mdd_object_type(obj)))
 		GOTO(out, rc = 0);
 
-        mdo_ref_add(env, obj, th);
-        mdd_orphan_ref_add(env, mdd, th);
+	mdo_ref_add(env, obj, th);
+	dt_ref_add(env, mdd->mdd_orphans, th);
 
-        /* try best to fixup directory, dont return errors
-         * from here */
-        if (!dt_try_as_dir(env, next))
+	/* try best to fixup directory, do not return errors from here */
+	if (!dt_try_as_dir(env, next))
 		GOTO(out, rc = 0);
-        next->do_index_ops->dio_delete(env, next,
-                                       (const struct dt_key *)dotdot,
-                                       th, BYPASS_CAPA);
 
-        next->do_index_ops->dio_insert(env, next,
-                                       (struct dt_rec *)lf_dor,
-                                       (const struct dt_key *)dotdot,
-                                       th, BYPASS_CAPA, 1);
+	dt_delete(env, next, (const struct dt_key *)dotdot, th);
+
+	rec->rec_fid = lf_dor;
+	rec->rec_type = S_IFDIR;
+	dt_insert(env, next, (const struct dt_rec *)rec,
+		  (const struct dt_key *)dotdot, th);
 
 out:
-        if (rc == 0)
-                obj->mod_flags |= ORPHAN_OBJ;
+	if (rc == 0)
+		obj->mod_flags |= ORPHAN_OBJ;
 
-        mdd_orphan_write_unlock(env, mdd);
+	dt_write_unlock(env, mdd->mdd_orphans);
 
-        RETURN(rc);
+	RETURN(rc);
 }
 
-int orph_declare_index_delete(const struct lu_env *env,
-                              struct mdd_object *obj,
-
-                              struct thandle *th)
+int mdd_orphan_declare_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;
+	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+	struct dt_key *key;
+	int rc;
 
-	key = orph_key_fill(env, mdo2fid(obj), ORPH_OP_UNLINK);
+	key = mdd_orphan_key_fill(env, mdo2fid(obj));
 
 	rc = dt_declare_delete(env, mdd->mdd_orphans, key, th);
-        if (rc)
-                return rc;
+	if (rc)
+		return rc;
+
+	if (!mdd_object_exists(obj))
+		return -ENOENT;
 
         rc = mdo_declare_ref_del(env, obj, th);
         if (rc)
@@ -272,99 +228,125 @@ int orph_declare_index_delete(const struct lu_env *env,
         return rc;
 }
 
-static int orph_index_delete(const struct lu_env *env,
-                             struct mdd_object *obj,
-                             __u32 op,
-                             struct thandle *th)
+/**
+ *  delete an orphan \a obj from orphan index.
+ *  \param obj file or directory.
+ *  \param th  transaction for index deletion and object destruction.
+ *
+ *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
+ *
+ *  \retval 0  success
+ *  \retval  -ve index operation error.
+ */
+int mdd_orphan_delete(const struct lu_env *env, struct mdd_object *obj,
+		      struct thandle *th)
 {
-        struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
-        struct dt_object *dor = mdd->mdd_orphans;
-        struct dt_key *key;
-        int rc;
+	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+	struct dt_object *dor = mdd->mdd_orphans;
+	struct dt_key *key;
+	int rc = 0;
 
-        ENTRY;
+	ENTRY;
 
-        LASSERT(mdd_write_locked(env, obj) != 0);
-        LASSERT(obj->mod_flags & ORPHAN_OBJ);
-        LASSERT(obj->mod_count == 0);
+	LASSERT(mdd_write_locked(env, obj) != 0);
+	LASSERT(obj->mod_flags & ORPHAN_OBJ);
+	LASSERT(obj->mod_count == 0);
 
-        LASSERT(dor);
+	LASSERT(dor);
 
-        key = orph_key_fill(env, mdo2fid(obj), op);
-        mdd_orphan_write_lock(env, mdd);
+	key = mdd_orphan_key_fill(env, mdo2fid(obj));
+	dt_write_lock(env, mdd->mdd_orphans, DT_TGT_ORPHAN);
 
-        rc = mdd_orphan_delete_obj(env, mdd, key, th);
+	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ORPHAN_DELETE))
+		goto ref_del;
 
-        if (rc == -ENOENT) {
-                key = orph_key_fill_18(env, mdo2fid(obj));
-                rc = mdd_orphan_delete_obj(env, mdd, key, th);
-        }
+	rc = dt_delete(env, mdd->mdd_orphans, key, th);
+	if (rc == -ENOENT) {
+		key = mdd_orphan_key_fill_20(env, mdo2fid(obj));
+		rc = dt_delete(env, mdd->mdd_orphans, key, th);
+	}
 
-        if (!rc) {
-                /* lov objects will be destroyed by caller */
-                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);
-                }
-                obj->mod_flags &= ~ORPHAN_OBJ;
-        } else {
-                CERROR("could not delete object: rc = %d\n",rc);
-        }
+ref_del:
+	if (!rc) {
+		/* lov objects will be destroyed by caller */
+		mdo_ref_del(env, obj, th);
+		if (S_ISDIR(mdd_object_type(obj))) {
+			mdo_ref_del(env, obj, th);
+			dt_ref_del(env, mdd->mdd_orphans, th);
+		}
+		obj->mod_flags &= ~ORPHAN_OBJ;
+	} else {
+		CERROR("%s: could not delete orphan object "DFID": rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name, PFID(mdo2fid(obj)), rc);
+	}
 
-        mdd_orphan_write_unlock(env, mdd);
-        RETURN(rc);
+	dt_write_unlock(env, mdd->mdd_orphans);
+	RETURN(rc);
 }
 
 
-static int orphan_object_destroy(const struct lu_env *env,
-                                 struct mdd_object *obj,
-                                 struct dt_key *key)
+static int mdd_orphan_destroy(const struct lu_env *env, struct mdd_object *obj,
+			      struct dt_key *key)
 {
-        struct thandle *th = NULL;
-        struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
-        int rc = 0;
-        ENTRY;
+	struct thandle *th = NULL;
+	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
+	bool orphan_exists = true;
+	int rc = 0;
+	ENTRY;
 
 	th = mdd_trans_create(env, mdd);
 	if (IS_ERR(th)) {
-		CERROR("Cannot get thandle\n");
-		RETURN(PTR_ERR(th));
+		rc = PTR_ERR(th);
+		if (rc != -EINPROGRESS)
+			CERROR("%s: cannot get orphan thandle: rc = %d\n",
+			       mdd2obd_dev(mdd)->obd_name, rc);
+		RETURN(rc);
 	}
 
-        rc = orph_declare_index_delete(env, obj, th);
-        if (rc)
-                GOTO(stop, rc);
-
-	rc = mdo_declare_destroy(env, obj, th);
-        if (rc)
-                GOTO(stop, rc);
-
-        rc = mdd_trans_start(env, mdd, th);
-        if (rc)
-                GOTO(stop, rc);
+	mdd_write_lock(env, obj, DT_TGT_CHILD);
+	rc = mdd_orphan_declare_delete(env, obj, th);
+	if (rc == -ENOENT)
+		orphan_exists = false;
+	else if (rc)
+		GOTO(unlock, rc);
+
+	if (orphan_exists) {
+		rc = mdo_declare_destroy(env, obj, th);
+		if (rc)
+			GOTO(unlock, rc);
+	}
 
-        mdd_write_lock(env, obj, MOR_TGT_CHILD);
-        if (likely(obj->mod_count == 0)) {
-                mdd_orphan_write_lock(env, mdd);
-                rc = mdd_orphan_delete_obj(env, mdd, key, th);
-		if (rc == 0) {
+	rc = mdd_trans_start(env, mdd, th);
+	if (rc)
+		GOTO(unlock, rc);
+
+	if (likely(obj->mod_count == 0)) {
+		dt_write_lock(env, mdd->mdd_orphans, DT_TGT_ORPHAN);
+		rc = dt_delete(env, mdd->mdd_orphans, key, th);
+		if (rc) {
+			CERROR("%s: could not delete orphan "DFID": rc = %d\n",
+			       mdd2obd_dev(mdd)->obd_name, PFID(mdo2fid(obj)),
+			       rc);
+		} else if (orphan_exists) {
 			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);
+				dt_ref_del(env, mdd->mdd_orphans, th);
 			}
 			rc = mdo_destroy(env, obj, th);
-		} else
-                        CERROR("could not delete object: rc = %d\n",rc);
-                mdd_orphan_write_unlock(env, mdd);
-        }
-        mdd_write_unlock(env, obj);
+		} else {
+			CWARN("%s: orphan %s "DFID" doesn't exist\n",
+			      mdd2obd_dev(mdd)->obd_name, (char *)key,
+			      PFID(mdo2fid(obj)));
+		}
+		dt_write_unlock(env, mdd->mdd_orphans);
+	}
+unlock:
+	mdd_write_unlock(env, obj);
 
-stop:
-        mdd_trans_stop(env, mdd, 0, th);
+	rc = mdd_trans_stop(env, mdd, 0, th);
 
-        RETURN(rc);
+	RETURN(rc);
 }
 
 /**
@@ -377,29 +359,27 @@ stop:
  * \retval 0   success
  * \retval -ve error
  */
-static int orph_key_test_and_del(const struct lu_env *env,
-                                 struct mdd_device *mdd,
-                                 struct lu_fid *lf,
-                                 struct dt_key *key)
+static int mdd_orphan_key_test_and_delete(const struct lu_env *env,
+					  struct mdd_device *mdd,
+					  struct lu_fid *lf, struct dt_key *key)
 {
-        struct mdd_object *mdo;
-        int rc;
+	struct mdd_object *mdo;
+	int rc;
 
-        mdo = mdd_object_find(env, mdd, lf);
+	mdo = mdd_object_find(env, mdd, lf);
 
-        if (IS_ERR(mdo))
-                return PTR_ERR(mdo);
+	if (IS_ERR(mdo))
+		return PTR_ERR(mdo);
 
-        rc = -EBUSY;
-        if (mdo->mod_count == 0) {
-                CDEBUG(D_HA, "Found orphan "DFID", delete it\n", PFID(lf));
-                rc = orphan_object_destroy(env, mdo, key);
-                if (rc) /* so replay-single.sh test_37 works */
-                        CERROR("%s: error unlinking orphan "DFID" from "
-                               "PENDING: rc = %d\n",
+	rc = -EBUSY;
+	if (mdo->mod_count == 0) {
+		CDEBUG(D_HA, "Found orphan "DFID", delete it\n", PFID(lf));
+		rc = mdd_orphan_destroy(env, mdo, key);
+		if (rc) /* below message checked in replay-single.sh test_37 */
+			CERROR("%s: error unlinking orphan "DFID": rc = %d\n",
 			       mdd2obd_dev(mdd)->obd_name, PFID(lf), rc);
         } else {
-                mdd_write_lock(env, mdo, MOR_TGT_CHILD);
+		mdd_write_lock(env, mdo, DT_TGT_CHILD);
                 if (likely(mdo->mod_count > 0)) {
                         CDEBUG(D_HA, "Found orphan "DFID" count %d, skip it\n",
                                PFID(lf), mdo->mod_count);
@@ -424,38 +404,38 @@ static int orph_key_test_and_del(const struct lu_env *env,
  * \retval 0   success
  * \retval -ve error
  */
-static int orph_index_iterate(const struct lu_env *env,
-			      struct mdd_generic_thread *thread)
+static int mdd_orphan_index_iterate(const struct lu_env *env,
+				    struct mdd_generic_thread *thread)
 {
 	struct mdd_device *mdd = (struct mdd_device *)thread->mgt_data;
 	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;
-        ENTRY;
-
-        iops = &dor->do_index_ops->dio_it;
-        it = iops->init(env, dor, LUDA_64BITHASH, BYPASS_CAPA);
-        if (IS_ERR(it)) {
-                rc = PTR_ERR(it);
-                CERROR("%s: cannot clean PENDING: rc = %d\n",
-		       mdd2obd_dev(mdd)->obd_name, rc);
-                GOTO(out, rc);
-        }
+	struct dt_it *it;
+	struct lu_fid fid;
+	int key_sz = 0;
+	int rc;
+	__u64 cookie;
+	ENTRY;
 
-        rc = iops->load(env, it, 0);
-        if (rc < 0)
-                GOTO(out_put, rc);
-        if (rc == 0) {
-                CERROR("%s: error loading iterator to clean PENDING\n",
-		       mdd2obd_dev(mdd)->obd_name);
-                /* Index contains no zero key? */
-                GOTO(out_put, rc = -EIO);
-        }
+	iops = &dor->do_index_ops->dio_it;
+	it = iops->init(env, dor, LUDA_64BITHASH);
+	if (IS_ERR(it)) {
+		rc = PTR_ERR(it);
+		CERROR("%s: cannot clean '%s': rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name, mdd_orphan_index_name, rc);
+		GOTO(out, rc);
+	}
+
+	rc = iops->load(env, it, 0);
+	if (rc < 0)
+		GOTO(out_put, rc);
+	if (rc == 0) {
+		CERROR("%s: error loading iterator to clean '%s'\n",
+		       mdd2obd_dev(mdd)->obd_name, mdd_orphan_index_name);
+		/* Index contains no zero key? */
+		GOTO(out_put, rc = -EIO);
+	}
 
 	do {
 		if (thread->mgt_abort)
@@ -475,16 +455,17 @@ static int orph_index_iterate(const struct lu_env *env,
 
 		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));
+			CERROR("%s: bad FID "DFID" cleaning '%s'\n",
+			       mdd2obd_dev(mdd)->obd_name, PFID(&fid),
+			       mdd_orphan_index_name);
 			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);
+		rc = mdd_orphan_key_test_and_delete(env, mdd, &fid,
+						(struct dt_key *)ent->lde_name);
 
 		/* after index delete reset iterator */
 		if (rc == 0)
@@ -520,18 +501,18 @@ out:
  *  \retval  -ve index operation error.
  *
  */
-int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
+int mdd_orphan_index_init(const struct lu_env *env, struct mdd_device *mdd)
 {
-	struct lu_fid		 fid;
-	struct dt_object	*d;
-	int			 rc = 0;
+	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 |
+				   mdd_orphan_index_name, S_IFDIR | S_IRUGO |
 				   S_IWUSR | S_IXUGO, &fid);
 	if (rc < 0)
 		RETURN(rc);
@@ -541,30 +522,30 @@ int orph_index_init(const struct lu_env *env, struct mdd_device *mdd)
 		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);
+		CERROR("%s: orphan dir '%s' is not an index: rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name, mdd_orphan_index_name, rc);
+		dt_object_put(env, d);
 		RETURN(-ENOTDIR);
 	}
 	mdd->mdd_orphans = d;
 	RETURN(0);
 }
 
-void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
+void mdd_orphan_index_fini(const struct lu_env *env, struct mdd_device *mdd)
 {
-        ENTRY;
-        if (mdd->mdd_orphans != NULL) {
-                lu_object_put(env, &mdd->mdd_orphans->do_lu);
-                mdd->mdd_orphans = NULL;
-        }
-        EXIT;
+	ENTRY;
+	if (mdd->mdd_orphans != NULL) {
+		dt_object_put(env, mdd->mdd_orphans);
+		mdd->mdd_orphans = NULL;
+	}
+	EXIT;
 }
 
-static int __mdd_orphan_cleanup(void *args)
+static int mdd_orphan_cleanup_thread(void *args)
 {
 	struct mdd_generic_thread *thread = (struct mdd_generic_thread *)args;
-	struct lu_env		  *env = NULL;
-	int			   rc;
+	struct lu_env *env = NULL;
+	int rc;
 	ENTRY;
 
 	complete(&thread->mgt_started);
@@ -577,7 +558,7 @@ static int __mdd_orphan_cleanup(void *args)
 	if (rc)
 		GOTO(out, rc);
 
-	rc = orph_index_iterate(env, thread);
+	rc = mdd_orphan_index_iterate(env, thread);
 
 	lu_env_fini(env);
 	GOTO(out, rc);
@@ -601,48 +582,16 @@ int mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
 	if (name == NULL)
 		goto out;
 
-	snprintf(name, MTI_NAME_MAXLEN, "orph_cleanup_%s",
-		 mdd2obd_dev(d)->obd_name);
+	snprintf(name, MTI_NAME_MAXLEN, "orph_%s", mdd2obd_dev(d)->obd_name);
 
-	rc = mdd_generic_thread_start(&d->mdd_orph_cleanup_thread,
-				      __mdd_orphan_cleanup, (void *)d, name);
+	rc = mdd_generic_thread_start(&d->mdd_orphan_cleanup_thread,
+				      mdd_orphan_cleanup_thread, d, name);
 out:
 	if (rc)
-		CERROR("%s: start orphan cleanup thread failed:%d\n",
+		CERROR("%s: start orphan cleanup thread failed: rc = %d\n",
 		       mdd2obd_dev(d)->obd_name, rc);
 	if (name)
 		OBD_FREE(name, MTI_NAME_MAXLEN);
-	return rc;
-}
-
-/**
- *  add an orphan \a obj to the orphan index.
- *  \param obj file or directory.
- *  \param th  transaction for index insert.
- *
- *  \pre obj nlink == 0 && obj->mod_count != 0
- *
- *  \retval 0  success
- *  \retval  -ve index operation error.
- */
-int __mdd_orphan_add(const struct lu_env *env,
-                     struct mdd_object *obj, struct thandle *th)
-{
-        return orph_index_insert(env, obj, ORPH_OP_UNLINK, th);
-}
 
-/**
- *  delete an orphan \a obj from orphan index.
- *  \param obj file or directory.
- *  \param th  transaction for index deletion and object destruction.
- *
- *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
- *
- *  \retval 0  success
- *  \retval  -ve index operation error.
- */
-int __mdd_orphan_del(const struct lu_env *env,
-                     struct mdd_object *obj, struct thandle *th)
-{
-        return orph_index_delete(env, obj, ORPH_OP_UNLINK, th);
+	return rc;
 }