From: wang di <di.wang@intel.com>
Date: Fri, 23 May 2014 08:44:33 +0000 (-0700)
Subject: LU-3531 mdt: delete striped directory
X-Git-Tag: 2.5.56~1
X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=370de927fc58fd3910fc527a00b5ff96da4a4278

LU-3531 mdt: delete striped directory

Add delete striped directory, it includes

1. enable sync log between MDTs, so slave objects will
be delete by unlink log, which is similar as deleting ost
object.

2. retrieve layout information of striped directory on MDT,
then lock all of the slave objects before unlink.

3. remove a few unnecessary cfs_size_round, because update_size
and update_buf_size already do size_around inside.

4. add sanity 300 for striped dir test

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: Ib461156bbff9e416ac9d2500a0b9491427542340
Reviewed-on: http://review.whamcloud.com/7445
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: John L. Hammond <john.hammond@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
---

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index cdd321a..6d6cecc 100644
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -278,6 +278,8 @@ enum dt_format_type dt_mode_to_dft(__u32 mode);
 
 typedef __u64 dt_obj_version_t;
 
+union ldlm_policy_data;
+
 /**
  * Per-dt-object operations.
  */
@@ -467,7 +469,11 @@ struct dt_object_operations {
 	int (*do_object_lock)(const struct lu_env *env, struct dt_object *dt,
 			      struct lustre_handle *lh,
 			      struct ldlm_enqueue_info *einfo,
-			      void *policy);
+			      union ldlm_policy_data *policy);
+
+	int (*do_object_unlock)(const struct lu_env *env, struct dt_object *dt,
+				struct ldlm_enqueue_info *einfo,
+				union ldlm_policy_data *policy);
 };
 
 /**
@@ -936,14 +942,25 @@ int local_object_unlink(const struct lu_env *env, struct dt_device *dt,
 static inline int dt_object_lock(const struct lu_env *env,
 				 struct dt_object *o, struct lustre_handle *lh,
 				 struct ldlm_enqueue_info *einfo,
-				 void *policy)
+				 union ldlm_policy_data *policy)
 {
-	LASSERT(o);
-	LASSERT(o->do_ops);
-	LASSERT(o->do_ops->do_object_lock);
+	LASSERT(o != NULL);
+	LASSERT(o->do_ops != NULL);
+	LASSERT(o->do_ops->do_object_lock != NULL);
 	return o->do_ops->do_object_lock(env, o, lh, einfo, policy);
 }
 
+static inline int dt_object_unlock(const struct lu_env *env,
+				   struct dt_object *o,
+				   struct ldlm_enqueue_info *einfo,
+				   union ldlm_policy_data *policy)
+{
+	LASSERT(o != NULL);
+	LASSERT(o->do_ops != NULL);
+	LASSERT(o->do_ops->do_object_unlock != NULL);
+	return o->do_ops->do_object_unlock(env, o, einfo, policy);
+}
+
 int dt_lookup_dir(const struct lu_env *env, struct dt_object *dir,
 		  const char *name, struct lu_fid *fid);
 
diff --git a/lustre/include/lustre_dlm.h b/lustre/include/lustre_dlm.h
index 98e03c9..60583f7 100644
--- a/lustre/include/lustre_dlm.h
+++ b/lustre/include/lustre_dlm.h
@@ -631,11 +631,13 @@ struct ldlm_flock {
         __u32 pid;
 };
 
-typedef union {
-        struct ldlm_extent l_extent;
-        struct ldlm_flock l_flock;
-        struct ldlm_inodebits l_inodebits;
-} ldlm_policy_data_t;
+union ldlm_policy_data {
+	struct ldlm_extent l_extent;
+	struct ldlm_flock l_flock;
+	struct ldlm_inodebits l_inodebits;
+};
+
+typedef union ldlm_policy_data ldlm_policy_data_t;
 
 void ldlm_convert_policy_to_wire(ldlm_type_t type,
                                  const ldlm_policy_data_t *lpolicy,
@@ -1053,6 +1055,7 @@ struct ldlm_enqueue_info {
 	void *ei_cb_cp;  /** lock completion callback */
 	void *ei_cb_gl;  /** lock glimpse callback */
 	void *ei_cbdata; /** Data to be passed into callbacks. */
+	unsigned int ei_enq_slave:1; /* whether enqueue slave stripes */
 };
 
 extern struct obd_ops ldlm_obd_ops;
diff --git a/lustre/include/lustre_fid.h b/lustre/include/lustre_fid.h
index cf8ba16..0cf2cc2 100644
--- a/lustre/include/lustre_fid.h
+++ b/lustre/include/lustre_fid.h
@@ -233,6 +233,7 @@ enum local_oid {
 	MDD_LOV_OBJ_OSEQ	= 4121UL,
 	LFSCK_NAMESPACE_OID     = 4122UL,
 	REMOTE_PARENT_DIR_OID	= 4123UL,
+	SLAVE_LLOG_CATALOGS_OID	= 4124UL,
 };
 
 static inline void lu_local_obj_fid(struct lu_fid *fid, __u32 oid)
diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index a780b7c..061c100 100644
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -230,7 +230,8 @@ int obd_llog_finish(struct obd_device *obd, int count);
 int llog_ioctl(const struct lu_env *env, struct llog_ctxt *ctxt, int cmd,
 	       struct obd_ioctl_data *data);
 int llog_catalog_list(const struct lu_env *env, struct dt_device *d,
-		      int count, struct obd_ioctl_data *data);
+		      int count, struct obd_ioctl_data *data,
+		      const struct lu_fid *fid);
 
 /* llog_net.c */
 int llog_initiator_connect(struct llog_ctxt *ctxt);
@@ -337,11 +338,11 @@ struct llog_handle {
 /* llog_osd.c */
 extern struct llog_operations llog_osd_ops;
 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
-			  int idx, int count,
-			  struct llog_catid *idarray);
+			  int idx, int count, struct llog_catid *idarray,
+			  const struct lu_fid *fid);
 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
-			  int idx, int count,
-			  struct llog_catid *idarray);
+			  int idx, int count, struct llog_catid *idarray,
+			  const struct lu_fid *fid);
 
 #define LLOG_CTXT_FLAG_UNINITIALIZED     0x00000001
 #define LLOG_CTXT_FLAG_STOP		 0x00000002
diff --git a/lustre/include/md_object.h b/lustre/include/md_object.h
index e85d956..82f6322 100644
--- a/lustre/include/md_object.h
+++ b/lustre/include/md_object.h
@@ -180,6 +180,7 @@ struct md_op_spec {
         const struct dt_index_features *sp_feat;
 };
 
+union ldlm_policy_data;
 /**
  * Operations implemented for each md object (both directory and leaf).
  */
@@ -255,7 +256,11 @@ struct md_object_operations {
 	int (*moo_object_lock)(const struct lu_env *env, struct md_object *obj,
 			       struct lustre_handle *lh,
 			       struct ldlm_enqueue_info *einfo,
-			       void *policy);
+			       union ldlm_policy_data *policy);
+	int (*moo_object_unlock)(const struct lu_env *env,
+				 struct md_object *obj,
+				 struct ldlm_enqueue_info *einfo,
+				 union ldlm_policy_data *policy);
 };
 
 /**
@@ -670,12 +675,21 @@ static inline int mo_object_lock(const struct lu_env *env,
 				 struct md_object *m,
 				 struct lustre_handle *lh,
 				 struct ldlm_enqueue_info *einfo,
-				 void *policy)
+				 union ldlm_policy_data *policy)
 {
 	LASSERT(m->mo_ops->moo_object_lock);
 	return m->mo_ops->moo_object_lock(env, m, lh, einfo, policy);
 }
 
+static inline int mo_object_unlock(const struct lu_env *env,
+				   struct md_object *m,
+				   struct ldlm_enqueue_info *einfo,
+				   union ldlm_policy_data *policy)
+{
+	LASSERT(m->mo_ops->moo_object_unlock);
+	return m->mo_ops->moo_object_unlock(env, m, einfo, policy);
+}
+
 static inline int mdo_lookup(const struct lu_env *env,
                              struct md_object *p,
                              const struct lu_name *lname,
diff --git a/lustre/llite/dir.c b/lustre/llite/dir.c
index 2b1d227..6a1e3c5 100644
--- a/lustre/llite/dir.c
+++ b/lustre/llite/dir.c
@@ -374,16 +374,6 @@ static int ll_dir_setdirstripe(struct inode *dir, struct lmv_user_md *lump,
 	if (unlikely(lump->lum_magic != LMV_USER_MAGIC))
 		RETURN(-EINVAL);
 
-	if (lump->lum_stripe_offset == (__u32)-1) {
-		int mdtidx;
-
-		mdtidx = ll_get_mdt_idx(dir);
-		if (mdtidx < 0)
-			RETURN(mdtidx);
-
-		lump->lum_stripe_offset = mdtidx;
-	}
-
 	CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p) name %s"
 	       "stripe_offset %d, stripe_count: %u\n",
 	       PFID(ll_inode2fid(dir)), dir, filename,
diff --git a/lustre/lmv/lmv_intent.c b/lustre/lmv/lmv_intent.c
index 0eae5a8..363eb3a 100644
--- a/lustre/lmv/lmv_intent.c
+++ b/lustre/lmv/lmv_intent.c
@@ -253,6 +253,12 @@ update:
 				if (req != NULL)
 					ptlrpc_req_finished(req);
 
+				if (it.d.lustre.it_lock_mode && lockh) {
+					ldlm_lock_decref(lockh,
+						 it.d.lustre.it_lock_mode);
+					it.d.lustre.it_lock_mode = 0;
+				}
+
 				GOTO(cleanup, rc = -EIO);
 			}
 
diff --git a/lustre/lmv/lmv_obd.c b/lustre/lmv/lmv_obd.c
index 4e1fb39..ee5f3a7 100644
--- a/lustre/lmv/lmv_obd.c
+++ b/lustre/lmv/lmv_obd.c
@@ -111,6 +111,9 @@ int lmv_name_to_stripe_index(enum lmv_hash_type hashtype,
 		return -EINVAL;
 	}
 
+	CDEBUG(D_INFO, "name %.*s hash_type %d idx %d\n", namelen, name,
+	       hashtype, idx);
+
 	LASSERT(idx < max_mdt_index);
 	return idx;
 }
@@ -1344,7 +1347,15 @@ static int lmv_placement_policy(struct obd_device *obd,
 		struct lmv_user_md *lum;
 
 		lum = op_data->op_data;
-		*mds = lum->lum_stripe_offset;
+
+		if (lum->lum_stripe_offset != (__u32)-1) {
+			*mds = lum->lum_stripe_offset;
+		} else {
+			/* -1 means default, which will be in the same MDT with
+			 * the stripe */
+			*mds = op_data->op_mds;
+			lum->lum_stripe_offset = op_data->op_mds;
+		}
 	} else {
 		/* Allocate new fid on target according to operation type and
 		 * parent home mds. */
@@ -1765,13 +1776,29 @@ static int lmv_close(struct obd_export *exp, struct md_op_data *op_data,
  * For striped-directory, it will locate MDT by name. And also
  * it will reset op_fid1 with the FID of the choosen stripe.
  **/
+struct lmv_tgt_desc *
+lmv_locate_target_for_name(struct lmv_obd *lmv, struct lmv_stripe_md *lsm,
+			   const char *name, int namelen, struct lu_fid *fid,
+			   mdsno_t *mds)
+{
+	struct lmv_tgt_desc	*tgt;
+	const struct lmv_oinfo	*oinfo;
+
+	oinfo = lsm_name_to_stripe_info(lsm, name, namelen);
+	*fid = oinfo->lmo_fid;
+	*mds = oinfo->lmo_mds;
+	tgt = lmv_get_target(lmv, *mds);
+
+	CDEBUG(D_INFO, "locate on mds %u "DFID"\n", *mds, PFID(fid));
+	return tgt;
+}
+
 struct lmv_tgt_desc
 *lmv_locate_mds(struct lmv_obd *lmv, struct md_op_data *op_data,
 		struct lu_fid *fid)
 {
 	struct lmv_stripe_md	*lsm = op_data->op_mea1;
 	struct lmv_tgt_desc	*tgt;
-	const struct lmv_oinfo	*oinfo;
 
 	if (lsm == NULL || lsm->lsm_md_stripe_count <= 1 ||
 	    op_data->op_namelen == 0) {
@@ -1783,15 +1810,9 @@ struct lmv_tgt_desc
 		return tgt;
 	}
 
-	oinfo = lsm_name_to_stripe_info(lsm, op_data->op_name,
-					op_data->op_namelen);
-	*fid = oinfo->lmo_fid;
-	op_data->op_mds = oinfo->lmo_mds;
-	tgt = lmv_get_target(lmv, op_data->op_mds);
-
-	CDEBUG(D_INFO, "locate on mds %u\n", op_data->op_mds);
-
-	return tgt;
+	return lmv_locate_target_for_name(lmv, lsm, op_data->op_name,
+					  op_data->op_namelen, fid,
+					  &op_data->op_mds);
 }
 
 int lmv_create(struct obd_export *exp, struct md_op_data *op_data,
@@ -2195,6 +2216,9 @@ static int lmv_rename(struct obd_export *exp, struct md_op_data *op_data,
 				      LCK_EX, MDS_INODELOCK_FULL,
 				      MF_MDC_CANCEL_FID4);
 
+	CDEBUG(D_INODE, DFID":m%d to "DFID"\n", PFID(&op_data->op_fid1),
+	       op_data->op_mds, PFID(&op_data->op_fid2));
+
 	if (rc == 0)
 		rc = md_rename(src_tgt->ltd_exp, op_data, old, oldlen,
 			       new, newlen, request);
@@ -2457,12 +2481,27 @@ static int lmv_unlink(struct obd_export *exp, struct md_op_data *op_data,
 		RETURN(rc);
 retry:
 	/* Send unlink requests to the MDT where the child is located */
-	if (likely(!fid_is_zero(&op_data->op_fid2)))
-		tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid2);
-	else
+	if (likely(!fid_is_zero(&op_data->op_fid2))) {
+		tgt = lmv_find_target(lmv, &op_data->op_fid2);
+		if (IS_ERR(tgt))
+			RETURN(PTR_ERR(tgt));
+
+		/* For striped dir, we need to locate the parent as well */
+		if (op_data->op_mea1 != NULL &&
+		    op_data->op_mea1->lsm_md_stripe_count > 1) {
+			LASSERT(op_data->op_name != NULL &&
+				op_data->op_namelen != 0);
+			lmv_locate_target_for_name(lmv, op_data->op_mea1,
+						   op_data->op_name,
+						   op_data->op_namelen,
+						   &op_data->op_fid1,
+						   &op_data->op_mds);
+		}
+	} else {
 		tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
-	if (IS_ERR(tgt))
-		RETURN(PTR_ERR(tgt));
+		if (IS_ERR(tgt))
+			RETURN(PTR_ERR(tgt));
+	}
 
 	op_data->op_fsuid = current_fsuid();
 	op_data->op_fsgid = current_fsgid();
@@ -2983,8 +3022,10 @@ int lmv_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
 	struct lmv_tgt_desc	*tgt = lmv->tgts[0];
 	ENTRY;
 
-	if (md->lmv != NULL)
+	if (md->lmv != NULL) {
 		lmv_free_memmd(md->lmv);
+		md->lmv = NULL;
+	}
 	if (tgt == NULL || tgt->ltd_exp == NULL)
 		RETURN(-EINVAL);
 	RETURN(md_free_lustre_md(lmv->tgts[0]->ltd_exp, md));
diff --git a/lustre/lod/lod_object.c b/lustre/lod/lod_object.c
index 5f69061..90f2da6 100644
--- a/lustre/lod/lod_object.c
+++ b/lustre/lod/lod_object.c
@@ -409,9 +409,9 @@ static int lod_attr_set(const struct lu_env *env,
 		if ((attr->la_valid & LA_ATIME &&
 		     attr->la_atime < la->la_atime) ||
 		    (attr->la_valid & LA_CTIME &&
-		     attr->la_atime < la->la_ctime) ||
+		     attr->la_ctime < la->la_ctime) ||
 		    (attr->la_valid & LA_MTIME &&
-		     attr->la_atime < la->la_mtime))
+		     attr->la_mtime < la->la_mtime))
 			setattr_time = true;
 
 		if (!setattr_time)
@@ -517,6 +517,7 @@ int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
 	struct lmv_mds_md_v1	*lmm1;
 	int			stripe_count;
 	int			lmm_size;
+	int			type = LU_SEQ_RANGE_ANY;
 	int			i;
 	int			rc;
 	__u32			mdtidx;
@@ -537,7 +538,7 @@ int lod_prep_lmv_md(const struct lu_env *env, struct dt_object *dt,
 	lmm1->lmv_stripe_count = cpu_to_le32(stripe_count);
 	lmm1->lmv_hash_type = cpu_to_le32(lo->ldo_dir_hash_type);
 	rc = lod_fld_lookup(env, lod, lu_object_fid(&dt->do_lu),
-			    &mdtidx, LU_SEQ_RANGE_MDT);
+			    &mdtidx, &type);
 	if (rc != 0)
 		RETURN(rc);
 
@@ -596,11 +597,11 @@ int lod_parse_dir_striping(const struct lu_env *env, struct lod_object *lo,
 	for (i = 1; i < le32_to_cpu(lmv1->lmv_stripe_count); i++) {
 		struct lod_tgt_desc	*tgt;
 		int			idx;
+		int			type = LU_SEQ_RANGE_ANY;
 		struct dt_object	*dto;
 
 		fid_le_to_cpu(fid, &lmv1->lmv_stripe_fids[i]);
-		rc = lod_fld_lookup(env, lod, fid,
-				    &idx, LU_SEQ_RANGE_MDT);
+		rc = lod_fld_lookup(env, lod, fid, &idx, &type);
 		if (rc != 0)
 			GOTO(out, rc);
 
@@ -1942,9 +1943,12 @@ static int lod_object_destroy(const struct lu_env *env,
 	/* destroy all underlying objects */
 	for (i = 0; i < lo->ldo_stripenr; i++) {
 		LASSERT(lo->ldo_stripe[i]);
-		rc = dt_destroy(env, lo->ldo_stripe[i], th);
-		if (rc)
-			break;
+		/* for striped directory, next == ldo_stripe[0] */
+		if (next != lo->ldo_stripe[i]) {
+			rc = dt_destroy(env, lo->ldo_stripe[i], th);
+			if (rc)
+				break;
+		}
 	}
 
 	RETURN(rc);
@@ -2003,19 +2007,137 @@ static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
 	return dt_object_sync(env, dt_object_child(dt));
 }
 
+struct lod_slave_locks	{
+	int			lsl_lock_count;
+	struct lustre_handle	lsl_handle[0];
+};
+
+static int lod_object_unlock_internal(const struct lu_env *env,
+				      struct dt_object *dt,
+				      struct ldlm_enqueue_info *einfo,
+				      ldlm_policy_data_t *policy)
+{
+	struct lod_object	*lo = lod_dt_obj(dt);
+	struct lod_slave_locks  *slave_locks = einfo->ei_cbdata;
+	int			rc = 0;
+	int			i;
+	ENTRY;
+
+	if (slave_locks == NULL)
+		RETURN(0);
+
+	for (i = 0; i < slave_locks->lsl_lock_count; i++) {
+		if (lustre_handle_is_used(&slave_locks->lsl_handle[i])) {
+			int	rc1;
+
+			einfo->ei_cbdata = &slave_locks->lsl_handle[i];
+			rc1 = dt_object_unlock(env, lo->ldo_stripe[i], einfo,
+					       policy);
+			if (rc1 < 0)
+				rc = rc == 0 ? rc1 : rc;
+		}
+	}
+
+	RETURN(rc);
+}
+
+static int lod_object_unlock(const struct lu_env *env, struct dt_object *dt,
+			     struct ldlm_enqueue_info *einfo,
+			     union ldlm_policy_data *policy)
+{
+	struct lod_object	*lo = lod_dt_obj(dt);
+	struct lod_slave_locks  *slave_locks = einfo->ei_cbdata;
+	int			slave_locks_size;
+	int			rc;
+	ENTRY;
+
+	if (slave_locks == NULL)
+		RETURN(0);
+
+	rc = lod_load_striping(env, lo);
+	if (rc != 0)
+		RETURN(rc);
+
+	/* Note: for remote lock for single stripe dir, MDT will cancel
+	 * the lock by lockh directly */
+	if (lo->ldo_stripenr == 0 && dt_object_remote(dt_object_child(dt)))
+		RETURN(0);
+
+	if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
+		RETURN(-ENOTDIR);
+
+	/* Only cancel slave lock for striped dir */
+	rc = lod_object_unlock_internal(env, dt, einfo, policy);
+
+	slave_locks_size = sizeof(*slave_locks) + slave_locks->lsl_lock_count *
+			   sizeof(slave_locks->lsl_handle[0]);
+	OBD_FREE(slave_locks, slave_locks_size);
+	einfo->ei_cbdata = NULL;
+
+	RETURN(rc);
+}
+
 static int lod_object_lock(const struct lu_env *env,
-			   struct dt_object *dt, struct lustre_handle *lh,
+			   struct dt_object *dt,
+			   struct lustre_handle *lh,
 			   struct ldlm_enqueue_info *einfo,
-			   void *policy)
+			   union ldlm_policy_data *policy)
 {
-	struct dt_object   *next = dt_object_child(dt);
-	int		 rc;
+	struct lod_object	*lo = lod_dt_obj(dt);
+	int			rc = 0;
+	int			i;
+	int			slave_locks_size;
+	struct lod_slave_locks	*slave_locks = NULL;
 	ENTRY;
 
-	/*
-	 * declare setattr on the local object
-	 */
-	rc = dt_object_lock(env, next, lh, einfo, policy);
+	/* remote object lock */
+	if (!einfo->ei_enq_slave) {
+		LASSERT(dt_object_remote(dt));
+		return dt_object_lock(env, dt_object_child(dt), lh, einfo,
+				      policy);
+	}
+
+	if (!S_ISDIR(dt->do_lu.lo_header->loh_attr))
+		RETURN(-ENOTDIR);
+
+	rc = lod_load_striping(env, lo);
+	if (rc != 0)
+		RETURN(rc);
+
+	/* No stripes */
+	if (lo->ldo_stripenr == 0)
+		RETURN(0);
+
+	slave_locks_size = sizeof(*slave_locks) + lo->ldo_stripenr *
+			   sizeof(slave_locks->lsl_handle[0]);
+	/* Freed in lod_object_unlock */
+	OBD_ALLOC(slave_locks, slave_locks_size);
+	if (slave_locks == NULL)
+		RETURN(-ENOMEM);
+	slave_locks->lsl_lock_count = lo->ldo_stripenr;
+
+	/* striped directory lock */
+	for (i = 0; i < lo->ldo_stripenr; i++) {
+		struct lustre_handle	lockh;
+
+		LASSERT(lo->ldo_stripe[i]);
+		rc = dt_object_lock(env, lo->ldo_stripe[i], &lockh, einfo,
+				    policy);
+		if (rc != 0)
+			GOTO(out, rc);
+
+		slave_locks->lsl_handle[i] = lockh;
+	}
+
+	einfo->ei_cbdata = slave_locks;
+
+out:
+	if (rc != 0 && slave_locks != NULL) {
+		einfo->ei_cbdata = slave_locks;
+		lod_object_unlock_internal(env, dt, einfo, policy);
+		OBD_FREE(slave_locks, slave_locks_size);
+		einfo->ei_cbdata = NULL;
+	}
 
 	RETURN(rc);
 }
@@ -2048,6 +2170,7 @@ struct dt_object_operations lod_obj_ops = {
 	.do_capa_get		= lod_capa_get,
 	.do_object_sync		= lod_object_sync,
 	.do_object_lock		= lod_object_lock,
+	.do_object_unlock	= lod_object_unlock,
 };
 
 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
diff --git a/lustre/mdd/mdd_object.c b/lustre/mdd/mdd_object.c
index 9e86b9e..56c3b2d 100644
--- a/lustre/mdd/mdd_object.c
+++ b/lustre/mdd/mdd_object.c
@@ -2039,13 +2039,22 @@ static int mdd_object_lock(const struct lu_env *env,
 			   struct md_object *obj,
 			   struct lustre_handle *lh,
 			   struct ldlm_enqueue_info *einfo,
-			   void *policy)
+			   ldlm_policy_data_t *policy)
 {
 	struct mdd_object *mdd_obj = md2mdd_obj(obj);
 	return dt_object_lock(env, mdd_object_child(mdd_obj), lh,
 			      einfo, policy);
 }
 
+static int mdd_object_unlock(const struct lu_env *env,
+			     struct md_object *obj,
+			     struct ldlm_enqueue_info *einfo,
+			     ldlm_policy_data_t *policy)
+{
+	struct mdd_object *mdd_obj = md2mdd_obj(obj);
+	return dt_object_unlock(env, mdd_object_child(mdd_obj), einfo, policy);
+}
+
 const struct md_object_operations mdd_obj_ops = {
 	.moo_permission		= mdd_permission,
 	.moo_attr_get		= mdd_attr_get,
@@ -2063,4 +2072,5 @@ const struct md_object_operations mdd_obj_ops = {
 	.moo_capa_get		= mdd_capa_get,
 	.moo_object_sync	= mdd_object_sync,
 	.moo_object_lock	= mdd_object_lock,
+	.moo_object_unlock	= mdd_object_unlock,
 };
diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 5aad150..90b2c5b 100644
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -483,7 +483,7 @@ void mdt_client_compatibility(struct mdt_thread_info *info)
 }
 
 static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o,
-			     char *name)
+			     const char *name)
 {
 	const struct lu_env *env = info->mti_env;
 	int rc;
@@ -522,7 +522,7 @@ static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o,
 }
 
 static int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o,
-			  struct md_attr *ma, char *name)
+			  struct md_attr *ma, const char *name)
 {
 	struct md_object *next = mdt_object_child(o);
 	struct lu_buf    *buf = &info->mti_buf;
@@ -2272,8 +2272,9 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
         RETURN(rc);
 }
 
-int mdt_md_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
-			void *data, int flag)
+/* Used for cross-MDT lock */
+int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
+			    void *data, int flag)
 {
 	struct lustre_handle lockh;
 	int		  rc;
@@ -2312,8 +2313,9 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti,
 	memset(einfo, 0, sizeof(*einfo));
 	einfo->ei_type = LDLM_IBITS;
 	einfo->ei_mode = mode;
-	einfo->ei_cb_bl = mdt_md_blocking_ast;
+	einfo->ei_cb_bl = mdt_remote_blocking_ast;
 	einfo->ei_cb_cp = ldlm_completion_ast;
+	einfo->ei_enq_slave = 0;
 
 	memset(policy, 0, sizeof(*policy));
 	policy->l_inodebits.bits = ibits;
@@ -5642,9 +5644,15 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
                 rc = mdt_ioc_version_get(mti, karg);
                 break;
         }
-	case OBD_IOC_CATLOGLIST:
-		rc = llog_catalog_list(&env, mdt->mdt_bottom, 0, karg);
+	case OBD_IOC_CATLOGLIST: {
+		struct mdt_thread_info *mti;
+
+		mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
+		lu_local_obj_fid(&mti->mti_tmp_fid1, LLOG_CATALOGS_OID);
+		rc = llog_catalog_list(&env, mdt->mdt_bottom, 0, karg,
+				       &mti->mti_tmp_fid1);
 		break;
+	 }
 	default:
 		rc = -EOPNOTSUPP;
 		CERROR("%s: Not supported cmd = %d, rc = %d\n",
diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index 79c6bcd..c3d4f7f 100644
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -745,6 +745,8 @@ enum {
 int mdt_get_info(struct tgt_session_info *tsi);
 int mdt_attr_get_complex(struct mdt_thread_info *info,
 			 struct mdt_object *o, struct md_attr *ma);
+int mdt_xattr_get(struct mdt_thread_info *info, struct mdt_object *o,
+		  struct md_attr *ma, const char *name);
 int mdt_ioepoch_open(struct mdt_thread_info *info, struct mdt_object *o,
                      int created);
 int mdt_object_is_som_enabled(struct mdt_object *mo);
@@ -791,6 +793,8 @@ extern struct lprocfs_vars lprocfs_mds_obd_vars[];
 int mdt_hsm_attr_set(struct mdt_thread_info *info, struct mdt_object *obj,
 		     const struct md_hsm *mh);
 
+int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
+			    void *data, int flag);
 /* mdt_idmap.c */
 int mdt_init_idmap(struct tgt_session_info *tsi);
 void mdt_cleanup_idmap(struct mdt_export_data *);
diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 98c2ad5..0afa5e8 100644
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -47,6 +47,7 @@
 #define DEBUG_SUBSYSTEM S_MDS
 
 #include "mdt_internal.h"
+#include <lustre_lmv.h>
 
 static inline void mdt_reint_init_ma(struct mdt_thread_info *info,
                                      struct md_attr *ma)
@@ -407,12 +408,61 @@ put_parent:
 	RETURN(rc);
 }
 
+static int mdt_unlock_slaves(struct mdt_thread_info *mti,
+			     struct mdt_object *obj, __u64 ibits,
+			     struct ldlm_enqueue_info *einfo)
+{
+	ldlm_policy_data_t	*policy = &mti->mti_policy;
+	int			rc;
+	ENTRY;
+
+	if (!S_ISDIR(obj->mot_header.loh_attr))
+		RETURN(0);
+
+	memset(policy, 0, sizeof(*policy));
+	policy->l_inodebits.bits = ibits;
+
+	rc = mo_object_unlock(mti->mti_env, mdt_object_child(obj), einfo,
+			      policy);
+	RETURN(rc);
+}
+
+/**
+ * Lock slave stripes if necessary, the lock handles of slave stripes
+ * will be stored in einfo->ei_cbdata.
+ **/
+static int mdt_lock_slaves(struct mdt_thread_info *mti, struct mdt_object *obj,
+			   ldlm_mode_t mode, __u64 ibits,
+			   struct ldlm_enqueue_info *einfo)
+{
+	ldlm_policy_data_t	*policy = &mti->mti_policy;
+	int			rc;
+	ENTRY;
+
+	if (!S_ISDIR(obj->mot_header.loh_attr))
+		RETURN(0);
+
+	memset(einfo, 0, sizeof(*einfo));
+	einfo->ei_type = LDLM_IBITS;
+	einfo->ei_mode = mode;
+	einfo->ei_cb_bl = mdt_remote_blocking_ast;
+	einfo->ei_cb_cp = ldlm_completion_ast;
+	einfo->ei_enq_slave = 1;
+	memset(policy, 0, sizeof(*policy));
+	policy->l_inodebits.bits = ibits;
+
+	rc = mo_object_lock(mti->mti_env, mdt_object_child(obj), NULL, einfo,
+			    policy);
+	RETURN(rc);
+}
+
 int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
                  struct md_attr *ma, int flags)
 {
         struct mdt_lock_handle  *lh;
         int do_vbr = ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID|LA_FLAGS);
         __u64 lockpart = MDS_INODELOCK_UPDATE;
+	struct ldlm_enqueue_info *einfo = &info->mti_einfo;
         int rc;
         ENTRY;
 
@@ -433,6 +483,10 @@ int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
         if (rc != 0)
                 RETURN(rc);
 
+	rc = mdt_lock_slaves(info, mo, LCK_EX, lockpart, einfo);
+	if (rc != 0)
+		GOTO(out_unlock, rc);
+
         if (mdt_object_exists(mo) == 0)
                 GOTO(out_unlock, rc = -ENOENT);
 
@@ -468,6 +522,7 @@ int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
 
         EXIT;
 out_unlock:
+	mdt_unlock_slaves(info, mo, lockpart, einfo);
         mdt_object_unlock(info, mo, lh, rc);
         return rc;
 }
@@ -724,8 +779,9 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
         struct mdt_object       *mc;
         struct mdt_lock_handle  *parent_lh;
         struct mdt_lock_handle  *child_lh;
-        int                      rc;
-	int			 no_name = 0;
+	struct ldlm_enqueue_info *einfo = &info->mti_einfo;
+	int			rc;
+	int			no_name = 0;
 	ENTRY;
 
 	DEBUG_REQ(D_INODE, req, "unlink "DFID"/"DNAME"", PFID(rr->rr_fid1),
@@ -739,6 +795,7 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
 
 	if (fid_is_obf(rr->rr_fid1) || fid_is_dot_lustre(rr->rr_fid1))
 		RETURN(-EPERM);
+
         /*
 	 * step 1: Found the parent.
          */
@@ -802,8 +859,6 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
 	if (fid_is_obf(child_fid) || fid_is_dot_lustre(child_fid))
 		GOTO(unlock_parent, rc = -EPERM);
 
-        mdt_reint_init_ma(info, ma);
-
 	/* We will lock the child regardless it is local or remote. No harm. */
 	mc = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
 	if (IS_ERR(mc))
@@ -874,17 +929,22 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
 			     MDT_CROSS_LOCK);
 	if (rc != 0)
 		GOTO(put_child, rc);
-
-        mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
-                       OBD_FAIL_MDS_REINT_UNLINK_WRITE);
-        /* save version when object is locked */
-        mdt_version_get_save(info, mc, 1);
 	/*
 	 * Now we can only make sure we need MA_INODE, in mdd layer, will check
 	 * whether need MA_LOV and MA_COOKIE.
 	 */
 	ma->ma_need = MA_INODE;
 	ma->ma_valid = 0;
+
+	rc = mdt_lock_slaves(info, mc, LCK_EX, MDS_INODELOCK_UPDATE, einfo);
+	if (rc != 0)
+		GOTO(unlock_child, rc);
+
+	mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
+		       OBD_FAIL_MDS_REINT_UNLINK_WRITE);
+	/* save version when object is locked */
+	mdt_version_get_save(info, mc, 1);
+
 	mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
 
 	mutex_lock(&mc->mot_lov_mutex);
@@ -919,8 +979,15 @@ static int mdt_reint_unlink(struct mdt_thread_info *info,
         }
 
         EXIT;
+
 unlock_child:
+	mdt_unlock_slaves(info, mc, MDS_INODELOCK_UPDATE, einfo);
 	mdt_object_unlock(info, mc, child_lh, rc);
+
+	/* Since we do not need reply md striped dir info to client, so
+	 * reset mti_big_lmm_used to avoid confusing mdt_fix_reply */
+	if (info->mti_big_lmm_used)
+		info->mti_big_lmm_used = 0;
 put_child:
 	mdt_object_put(info->mti_env, mc);
 unlock_parent:
diff --git a/lustre/obdclass/llog_ioctl.c b/lustre/obdclass/llog_ioctl.c
index 3391618..24471cc 100644
--- a/lustre/obdclass/llog_ioctl.c
+++ b/lustre/obdclass/llog_ioctl.c
@@ -429,7 +429,8 @@ out_close:
 EXPORT_SYMBOL(llog_ioctl);
 
 int llog_catalog_list(const struct lu_env *env, struct dt_device *d,
-		      int count, struct obd_ioctl_data *data)
+		      int count, struct obd_ioctl_data *data,
+		      const struct lu_fid *fid)
 {
 	int			 size, i;
 	struct llog_catid	*idarray;
@@ -440,7 +441,7 @@ int llog_catalog_list(const struct lu_env *env, struct dt_device *d,
 	ENTRY;
 
 	if (count == 0) { /* get total number of logs */
-		rc = llog_osd_get_cat_list(env, d, 0, 0, NULL);
+		rc = llog_osd_get_cat_list(env, d, 0, 0, NULL, fid);
 		if (rc < 0)
 			RETURN(rc);
 		count = rc;
@@ -452,7 +453,7 @@ int llog_catalog_list(const struct lu_env *env, struct dt_device *d,
 	if (!idarray)
 		RETURN(-ENOMEM);
 
-	rc = llog_osd_get_cat_list(env, d, 0, count, idarray);
+	rc = llog_osd_get_cat_list(env, d, 0, count, idarray, fid);
 	if (rc)
 		GOTO(out, rc);
 
diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index d9434d7..a47d709 100644
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -1169,7 +1169,8 @@ EXPORT_SYMBOL(llog_osd_ops);
 
 /* reads the catalog list */
 int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
-			  int idx, int count, struct llog_catid *idarray)
+			  int idx, int count, struct llog_catid *idarray,
+			  const struct lu_fid *fid)
 {
 	struct llog_thread_info	*lgi = llog_info(env);
 	struct dt_object	*o = NULL;
@@ -1183,8 +1184,7 @@ int llog_osd_get_cat_list(const struct lu_env *env, struct dt_device *d,
 	size = sizeof(*idarray) * count;
 	lgi->lgi_off = idx *  sizeof(*idarray);
 
-	lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
-
+	lgi->lgi_fid = *fid;
 	o = dt_locate(env, d, &lgi->lgi_fid);
 	if (IS_ERR(o))
 		RETURN(PTR_ERR(o));
@@ -1263,7 +1263,8 @@ EXPORT_SYMBOL(llog_osd_get_cat_list);
 
 /* writes the cat list */
 int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
-			  int idx, int count, struct llog_catid *idarray)
+			  int idx, int count, struct llog_catid *idarray,
+			  const struct lu_fid *fid)
 {
 	struct llog_thread_info	*lgi = llog_info(env);
 	struct dt_object	*o = NULL;
@@ -1277,8 +1278,7 @@ int llog_osd_put_cat_list(const struct lu_env *env, struct dt_device *d,
 
 	size = sizeof(*idarray) * count;
 	lgi->lgi_off = idx * sizeof(*idarray);
-
-	lu_local_obj_fid(&lgi->lgi_fid, LLOG_CATALOGS_OID);
+	lgi->lgi_fid = *fid;
 
 	o = dt_locate(env, d, &lgi->lgi_fid);
 	if (IS_ERR(o))
diff --git a/lustre/osd-ldiskfs/osd_scrub.c b/lustre/osd-ldiskfs/osd_scrub.c
index 9a5572f..e558163 100644
--- a/lustre/osd-ldiskfs/osd_scrub.c
+++ b/lustre/osd-ldiskfs/osd_scrub.c
@@ -1411,6 +1411,10 @@ static const struct osd_lf_map osd_lf_maps[] = {
 	{ "LAST_GROUP", { FID_SEQ_LOCAL_FILE, OFD_LAST_GROUP_OID, 0 },
 		OLF_SHOW_NAME, NULL, NULL },
 
+	/* SLAVE_LOG, llog for destroy slave stripes of striped dir */
+	{ "SLAVE_LOG", { FID_SEQ_LOCAL_FILE, SLAVE_LLOG_CATALOGS_OID, 0 },
+	       OLF_SHOW_NAME, NULL, NULL },
+
 	/* lost+found */
 	{ "lost+found", { 0, 0, 0 }, OLF_SCAN_SUBITEMS | OLF_NO_OI,
 		osd_ios_general_scan, osd_ios_lf_fill },
diff --git a/lustre/osp/osp_dev.c b/lustre/osp/osp_dev.c
index 41272f0..7f3311c 100644
--- a/lustre/osp/osp_dev.c
+++ b/lustre/osp/osp_dev.c
@@ -347,10 +347,9 @@ static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
 
 	rc = osp_disconnect(d);
 
-	if (!d->opd_connect_mdt) {
-		/* stop sync thread */
-		osp_sync_fini(d);
+	osp_sync_fini(d);
 
+	if (!d->opd_connect_mdt) {
 		/* stop precreate thread */
 		osp_precreate_fini(d);
 
@@ -709,17 +708,18 @@ static int osp_init0(const struct lu_env *env, struct osp_device *m,
 		rc = osp_init_precreate(m);
 		if (rc)
 			GOTO(out_last_used, rc);
-		/*
-		 * Initialize synhronization mechanism taking
-		 * care of propogating changes to OST in near
-		 * transactional manner.
-		 */
-		rc = osp_sync_init(env, m);
-		if (rc)
-			GOTO(out_precreat, rc);
 	}
 
 	/*
+	 * Initialize synhronization mechanism taking
+	 * care of propogating changes to OST in near
+	 * transactional manner.
+	 */
+	rc = osp_sync_init(env, m);
+	if (rc)
+		GOTO(out_precreat, rc);
+
+	/*
 	 * Initiate connect to OST
 	 */
 	ll_generate_random_uuid(uuid);
@@ -735,9 +735,8 @@ static int osp_init0(const struct lu_env *env, struct osp_device *m,
 	RETURN(0);
 
 out:
-	if (!m->opd_connect_mdt)
-		/* stop sync thread */
-		osp_sync_fini(m);
+	/* stop sync thread */
+	osp_sync_fini(m);
 out_precreat:
 	/* stop precreate thread */
 	if (!m->opd_connect_mdt)
diff --git a/lustre/osp/osp_md_object.c b/lustre/osp/osp_md_object.c
index 39fcb66..fc468d3 100644
--- a/lustre/osp/osp_md_object.c
+++ b/lustre/osp/osp_md_object.c
@@ -610,23 +610,22 @@ static int osp_md_object_lock(const struct lu_env *env,
 			      struct dt_object *dt,
 			      struct lustre_handle *lh,
 			      struct ldlm_enqueue_info *einfo,
-			      void *policy)
+			      ldlm_policy_data_t *policy)
 {
-	struct osp_thread_info *info = osp_env_info(env);
-	struct ldlm_res_id     *res_id = &info->osi_resid;
-	struct dt_device       *dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
-	struct osp_device      *osp = dt2osp_dev(dt_dev);
-	struct ptlrpc_request  *req = NULL;
-	int                     rc = 0;
-	__u64                   flags = 0;
-	ldlm_mode_t             mode;
+	struct osp_thread_info	*info = osp_env_info(env);
+	struct ldlm_res_id	*res_id = &info->osi_resid;
+	struct dt_device	*dt_dev = lu2dt_dev(dt->do_lu.lo_dev);
+	struct osp_device	*osp = dt2osp_dev(dt_dev);
+	struct ptlrpc_request	*req;
+	int			rc = 0;
+	__u64			flags = 0;
+	ldlm_mode_t		mode;
 
 	fid_build_reg_res_name(lu_object_fid(&dt->do_lu), res_id);
 
 	mode = ldlm_lock_match(osp->opd_obd->obd_namespace,
 			       LDLM_FL_BLOCK_GRANTED, res_id,
-			       einfo->ei_type,
-			       (ldlm_policy_data_t *)policy,
+			       einfo->ei_type, policy,
 			       einfo->ei_mode, lh, 0);
 	if (mode > 0)
 		return ELDLM_OK;
@@ -644,6 +643,19 @@ static int osp_md_object_lock(const struct lu_env *env,
 	return rc == ELDLM_OK ? 0 : -EIO;
 }
 
+static int osp_md_object_unlock(const struct lu_env *env,
+				struct dt_object *dt,
+				struct ldlm_enqueue_info *einfo,
+				ldlm_policy_data_t *policy)
+{
+	struct lustre_handle	*lockh = einfo->ei_cbdata;
+
+	/* unlock finally */
+	ldlm_lock_decref(lockh, einfo->ei_mode);
+
+	return 0;
+}
+
 struct dt_object_operations osp_md_obj_ops = {
 	.do_read_lock         = osp_md_object_read_lock,
 	.do_write_lock        = osp_md_object_write_lock,
@@ -667,4 +679,5 @@ struct dt_object_operations osp_md_obj_ops = {
 	.do_xattr_set         = osp_xattr_set,
 	.do_index_try         = osp_md_index_try,
 	.do_object_lock       = osp_md_object_lock,
+	.do_object_unlock     = osp_md_object_unlock,
 };
diff --git a/lustre/osp/osp_sync.c b/lustre/osp/osp_sync.c
index 493fbfe..9573273 100644
--- a/lustre/osp/osp_sync.c
+++ b/lustre/osp/osp_sync.c
@@ -348,8 +348,6 @@ static int osp_sync_interpret(const struct lu_env *env,
 	       rc, (unsigned) req->rq_transno);
 	LASSERT(rc || req->rq_transno);
 
-	LASSERT(d->opd_pre != NULL);
-
 	if (rc == -ENOENT) {
 		/*
 		 * we tried to destroy object or update attributes,
@@ -386,7 +384,8 @@ static int osp_sync_interpret(const struct lu_env *env,
 		}
 
 		wake_up(&d->opd_syn_waitq);
-	} else if (unlikely(d->opd_pre_status == -ENOSPC)) {
+	} else if (d->opd_pre != NULL &&
+		   unlikely(d->opd_pre_status == -ENOSPC)) {
 		/*
 		 * if current status is -ENOSPC (lack of free space on OST)
 		 * then we should poll OST immediately once object destroy
@@ -524,31 +523,100 @@ static int osp_sync_new_unlink_job(struct osp_device *d,
 	RETURN(0);
 }
 
-static int osp_sync_new_unlink64_job(struct osp_device *d,
+static int osp_prep_unlink_update_req(const struct lu_env *env,
+				      struct osp_device *osp,
+				      struct llog_handle *llh,
+				      struct llog_rec_hdr *h,
+				      struct ptlrpc_request **reqp)
+{
+	struct llog_unlink64_rec	*rec = (struct llog_unlink64_rec *)h;
+	struct update_request		*update = NULL;
+	struct ptlrpc_request		*req;
+	const char			*buf;
+	struct llog_cookie		lcookie;
+	int				size;
+	int				rc;
+	ENTRY;
+
+	update = out_create_update_req(&osp->opd_dt_dev);
+	if (IS_ERR(update))
+		RETURN(PTR_ERR(update));
+
+	/* This can only happens for unlink slave directory, so decrease
+	 * ref for ".." and "." */
+	rc = out_insert_update(env, update, OBJ_REF_DEL, &rec->lur_fid, 0,
+			       NULL, NULL);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	rc = out_insert_update(env, update, OBJ_REF_DEL, &rec->lur_fid, 0,
+			       NULL, NULL);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	lcookie.lgc_lgl = llh->lgh_id;
+	lcookie.lgc_subsys = LLOG_MDS_OST_ORIG_CTXT;
+	lcookie.lgc_index = h->lrh_index;
+	size = sizeof(lcookie);
+	buf = (const char *)&lcookie;
+
+	rc = out_insert_update(env, update, OBJ_DESTROY, &rec->lur_fid, 1,
+			       &size, &buf);
+	if (rc != 0)
+		GOTO(out, rc);
+
+	rc = out_prep_update_req(env, osp->opd_obd->u.cli.cl_import,
+				 update->ur_buf, UPDATE_BUFFER_SIZE, &req);
+
+	INIT_LIST_HEAD(&req->rq_exp_list);
+	req->rq_svc_thread = (void *)OSP_JOB_MAGIC;
+
+	req->rq_interpret_reply = osp_sync_interpret;
+	req->rq_commit_cb = osp_sync_request_commit_cb;
+	req->rq_cb_data = osp;
+
+	ptlrpc_request_set_replen(req);
+	*reqp = req;
+out:
+	if (update != NULL)
+		out_destroy_update_req(update);
+
+	RETURN(rc);
+}
+
+static int osp_sync_new_unlink64_job(const struct lu_env *env,
+				     struct osp_device *d,
 				     struct llog_handle *llh,
 				     struct llog_rec_hdr *h)
 {
 	struct llog_unlink64_rec	*rec = (struct llog_unlink64_rec *)h;
-	struct ptlrpc_request		*req;
+	struct ptlrpc_request		*req = NULL;
 	struct ost_body			*body;
 	int				 rc;
 
 	ENTRY;
 	LASSERT(h->lrh_type == MDS_UNLINK64_REC);
 
-	req = osp_sync_new_job(d, llh, h, OST_DESTROY, &RQF_OST_DESTROY);
-	if (IS_ERR(req))
-		RETURN(PTR_ERR(req));
-
-	body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
-	if (body == NULL)
-		RETURN(-EFAULT);
-	rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
-	if (rc < 0)
-		RETURN(rc);
-	body->oa.o_misc = rec->lur_count;
-	body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID | OBD_MD_FLOBJCOUNT;
+	if (d->opd_connect_mdt) {
+		rc = osp_prep_unlink_update_req(env, d, llh, h, &req);
+		if (rc != 0)
+			RETURN(rc);
+	} else {
+		req = osp_sync_new_job(d, llh, h, OST_DESTROY,
+				       &RQF_OST_DESTROY);
+		if (IS_ERR(req))
+			RETURN(PTR_ERR(req));
 
+		body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
+		if (body == NULL)
+			RETURN(-EFAULT);
+		rc = fid_to_ostid(&rec->lur_fid, &body->oa.o_oi);
+		if (rc < 0)
+			RETURN(rc);
+		body->oa.o_misc = rec->lur_count;
+		body->oa.o_valid = OBD_MD_FLGROUP | OBD_MD_FLID |
+				   OBD_MD_FLOBJCOUNT;
+	}
 	osp_sync_send_new_rpc(d, req);
 	RETURN(0);
 }
@@ -601,7 +669,7 @@ static int osp_sync_process_record(const struct lu_env *env,
 		rc = osp_sync_new_unlink_job(d, llh, rec);
 		break;
 	case MDS_UNLINK64_REC:
-		rc = osp_sync_new_unlink64_job(d, llh, rec);
+		rc = osp_sync_new_unlink64_job(env, d, llh, rec);
 		break;
 	case MDS_SETATTR64_REC:
 		rc = osp_sync_new_setattr_job(d, llh, rec);
@@ -666,8 +734,7 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	 * notice: we do this upon commit as well because some backends
 	 * (like DMU) do not release space right away.
 	 */
-	LASSERT(d->opd_pre != NULL);
-	if (unlikely(d->opd_pre_status == -ENOSPC))
+	if (d->opd_pre != NULL && unlikely(d->opd_pre_status == -ENOSPC))
 		osp_statfs_need_now(d);
 
 	/*
@@ -689,17 +756,35 @@ static void osp_sync_process_committed(const struct lu_env *env,
 	spin_unlock(&d->opd_syn_lock);
 
 	cfs_list_for_each_entry_safe(req, tmp, &list, rq_exp_list) {
+		struct llog_cookie *lcookie = NULL;
+
 		LASSERT(req->rq_svc_thread == (void *) OSP_JOB_MAGIC);
 		cfs_list_del_init(&req->rq_exp_list);
 
-		body = req_capsule_client_get(&req->rq_pill, &RMF_OST_BODY);
-		LASSERT(body);
-
+		if (d->opd_connect_mdt) {
+			struct update_buf *ubuf;
+			struct update *update;
+			ubuf = req_capsule_client_get(&req->rq_pill,
+						      &RMF_UPDATE);
+			LASSERT(ubuf != NULL &&
+				ubuf->ub_magic == UPDATE_BUFFER_MAGIC);
+			/* 1st/2nd is for decref . and .., 3rd one is for
+			 * destroy, where the log cookie is stored.
+			 * See osp_prep_unlink_update_req */
+			update = update_buf_get(ubuf, 2, NULL);
+			LASSERT(update != NULL);
+			lcookie = update_param_buf(update, 0, NULL);
+			LASSERT(lcookie != NULL);
+		} else {
+			body = req_capsule_client_get(&req->rq_pill,
+						      &RMF_OST_BODY);
+			LASSERT(body);
+			lcookie = &body->oa.o_lcookie;
+		}
 		/* import can be closing, thus all commit cb's are
 		 * called we can check committness directly */
 		if (req->rq_transno <= imp->imp_peer_committed_transno) {
-			rc = llog_cat_cancel_records(env, llh, 1,
-						     &body->oa.o_lcookie);
+			rc = llog_cat_cancel_records(env, llh, 1, lcookie);
 			if (rc)
 				CERROR("%s: can't cancel record: %d\n",
 				       obd->obd_name, rc);
@@ -903,11 +988,12 @@ out:
 
 static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 {
-	struct osp_thread_info *osi = osp_env_info(env);
-	struct llog_handle     *lgh = NULL;
-	struct obd_device      *obd = d->opd_obd;
-	struct llog_ctxt       *ctxt;
-	int                     rc;
+	struct osp_thread_info	*osi = osp_env_info(env);
+	struct lu_fid		*fid = &osi->osi_fid;
+	struct llog_handle	*lgh = NULL;
+	struct obd_device	*obd = d->opd_obd;
+	struct llog_ctxt	*ctxt;
+	int			rc;
 
 	ENTRY;
 
@@ -919,8 +1005,13 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 	OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
 	obd->obd_lvfs_ctxt.dt = d->opd_storage;
 
+	if (d->opd_connect_mdt)
+		lu_local_obj_fid(fid, SLAVE_LLOG_CATALOGS_OID);
+	else
+		lu_local_obj_fid(fid, LLOG_CATALOGS_OID);
+
 	rc = llog_osd_get_cat_list(env, d->opd_storage, d->opd_index, 1,
-				   &osi->osi_cid);
+				   &osi->osi_cid, fid);
 	if (rc) {
 		CERROR("%s: can't get id from catalogs: rc = %d\n",
 		       obd->obd_name, rc);
@@ -965,7 +1056,7 @@ static int osp_sync_llog_init(const struct lu_env *env, struct osp_device *d)
 		GOTO(out_close, rc);
 
 	rc = llog_osd_put_cat_list(env, d->opd_storage, d->opd_index, 1,
-				   &osi->osi_cid);
+				   &osi->osi_cid, fid);
 	if (rc)
 		GOTO(out_close, rc);
 
diff --git a/lustre/target/out_handler.c b/lustre/target/out_handler.c
index 0f58654..1d65bed 100644
--- a/lustre/target/out_handler.c
+++ b/lustre/target/out_handler.c
@@ -1352,7 +1352,7 @@ next:
 		lu_object_put(env, &dt_obj->do_lu);
 		if (rc < 0)
 			GOTO(out, rc);
-		off += cfs_size_round(update_size(update));
+		off += update_size(update);
 	}
 out:
 	rc1 = out_tx_end(env, ta);
diff --git a/lustre/target/out_lib.c b/lustre/target/out_lib.c
index 8ba1fd6..a335050 100644
--- a/lustre/target/out_lib.c
+++ b/lustre/target/out_lib.c
@@ -210,8 +210,7 @@ int out_insert_update(const struct lu_env *env, struct update_request *update,
 	int                   update_length;
 	ENTRY;
 
-	obj_update = (struct update *)((char *)ubuf +
-		      cfs_size_round(update_buf_size(ubuf)));
+	obj_update = (struct update *)((char *)ubuf + update_buf_size(ubuf));
 
 	/* Check update size to make sure it can fit into the buffer */
 	update_length = cfs_size_round(offsetof(struct update,
diff --git a/lustre/tests/racer/dir_remote.sh b/lustre/tests/racer/dir_remote.sh
index dbbaa19..b2d28fb 100755
--- a/lustre/tests/racer/dir_remote.sh
+++ b/lustre/tests/racer/dir_remote.sh
@@ -9,7 +9,7 @@ while /bin/true ; do
 	file=$((RANDOM % MAX))
 	mdt_idx=$((RANDOM % MDTCOUNT))
 	mkdir -p $DIR
-	lfs mkdir -i $mdt_idx $DIR/$remote_dir > /dev/null 2>&1
+	lfs mkdir -i$mdt_idx -c$MDTCOUNT $DIR/$remote_dir 2> /dev/null
 	echo "abcd" > $DIR/$remote_dir/$file 2> /dev/null
-	$LFS getdirstripe $DIR/$remote_dir > /dev/null 2>&1
+	$LFS getdirstripe $DIR/$remote_dir 2> /dev/null
 done
diff --git a/lustre/tests/sanity.sh b/lustre/tests/sanity.sh
index f580097..4a0ebee 100644
--- a/lustre/tests/sanity.sh
+++ b/lustre/tests/sanity.sh
@@ -792,9 +792,9 @@ test_24c() {
 run_test 24c "rename directory to non-existent target"
 
 test_24d() {
-	test_mkdir $DIR/$tdir
-	test_mkdir $DIR/$tdir/d$testnum.1
-	test_mkdir $DIR/$tdir/d$testnum.2
+	test_mkdir -c1 $DIR/$tdir
+	test_mkdir -c1 $DIR/$tdir/d$testnum.1
+	test_mkdir -c1 $DIR/$tdir/d$testnum.2
 	mrename $DIR/$tdir/d$testnum.1 $DIR/$tdir/d$testnum.2
 	$CHECKSTAT -a $DIR/$tdir/d$testnum.1 || error "d$testnum.1 exists"
 	$CHECKSTAT -t dir $DIR/$tdir/d$testnum.2 || error "d$testnum.2 not dir"
@@ -833,10 +833,10 @@ test_24g() {
 run_test 24g "mkdir .../R7{a,b}/d; mv .../R7a/d .../R7b/e ======"
 
 test_24h() {
-	test_mkdir $DIR/R8a
-	test_mkdir $DIR/R8b
-	test_mkdir $DIR/R8a/d
-	test_mkdir $DIR/R8b/e
+	test_mkdir -c1 $DIR/R8a
+	test_mkdir -c1 $DIR/R8b
+	test_mkdir -c1 $DIR/R8a/d
+	test_mkdir -c1 $DIR/R8b/e
 	mrename $DIR/R8a/d $DIR/R8b/e
 	$CHECKSTAT -a $DIR/R8a/d || error
 	$CHECKSTAT -t dir $DIR/R8b/e || error
@@ -2044,76 +2044,74 @@ test_31f() { # bug 4554
 run_test 31f "remove of open directory with open-unlink file ==="
 
 test_31g() {
-        echo "-- cross directory link --"
-        test_mkdir $DIR/d31ga
-        test_mkdir $DIR/d31gb
-        touch $DIR/d31ga/f
-        ln $DIR/d31ga/f $DIR/d31gb/g
-        $CHECKSTAT -t file $DIR/d31ga/f || error "source"
-        [ `stat -c%h $DIR/d31ga/f` == '2' ] || error "source nlink"
-        $CHECKSTAT -t file $DIR/d31gb/g || error "target"
-        [ `stat -c%h $DIR/d31gb/g` == '2' ] || error "target nlink"
+	echo "-- cross directory link --"
+	test_mkdir -c1 $DIR/${tdir}ga
+	test_mkdir -c1 $DIR/${tdir}gb
+	touch $DIR/${tdir}ga/f
+	ln $DIR/${tdir}ga/f $DIR/${tdir}gb/g
+	$CHECKSTAT -t file $DIR/${tdir}ga/f || error "source"
+	[ `stat -c%h $DIR/${tdir}ga/f` == '2' ] || error "source nlink"
+	$CHECKSTAT -t file $DIR/${tdir}gb/g || error "target"
+	[ `stat -c%h $DIR/${tdir}gb/g` == '2' ] || error "target nlink"
 }
 run_test 31g "cross directory link==============="
 
 test_31h() {
-        echo "-- cross directory link --"
-        test_mkdir $DIR/d31h
-        test_mkdir $DIR/d31h/dir
-        touch $DIR/d31h/f
-        ln $DIR/d31h/f $DIR/d31h/dir/g
-        $CHECKSTAT -t file $DIR/d31h/f || error "source"
-        [ `stat -c%h $DIR/d31h/f` == '2' ] || error "source nlink"
-        $CHECKSTAT -t file $DIR/d31h/dir/g || error "target"
-        [ `stat -c%h $DIR/d31h/dir/g` == '2' ] || error "target nlink"
+	echo "-- cross directory link --"
+	test_mkdir -c1 $DIR/${tdir}
+	test_mkdir -c1 $DIR/${tdir}/dir
+	touch $DIR/${tdir}/f
+	ln $DIR/${tdir}/f $DIR/${tdir}/dir/g
+	$CHECKSTAT -t file $DIR/${tdir}/f || error "source"
+	[ `stat -c%h $DIR/${tdir}/f` == '2' ] || error "source nlink"
+	$CHECKSTAT -t file $DIR/${tdir}/dir/g || error "target"
+	[ `stat -c%h $DIR/${tdir}/dir/g` == '2' ] || error "target nlink"
 }
 run_test 31h "cross directory link under child==============="
 
 test_31i() {
-        echo "-- cross directory link --"
-        test_mkdir $DIR/d31i
-        test_mkdir $DIR/d31i/dir
-        touch $DIR/d31i/dir/f
-        ln $DIR/d31i/dir/f $DIR/d31i/g
-        $CHECKSTAT -t file $DIR/d31i/dir/f || error "source"
-        [ `stat -c%h $DIR/d31i/dir/f` == '2' ] || error "source nlink"
-        $CHECKSTAT -t file $DIR/d31i/g || error "target"
-        [ `stat -c%h $DIR/d31i/g` == '2' ] || error "target nlink"
+	echo "-- cross directory link --"
+	test_mkdir -c1 $DIR/$tdir
+	test_mkdir -c1 $DIR/$tdir/dir
+	touch $DIR/$tdir/dir/f
+	ln $DIR/$tdir/dir/f $DIR/$tdir/g
+	$CHECKSTAT -t file $DIR/$tdir/dir/f || error "source"
+	[ `stat -c%h $DIR/$tdir/dir/f` == '2' ] || error "source nlink"
+	$CHECKSTAT -t file $DIR/$tdir/g || error "target"
+	[ `stat -c%h $DIR/$tdir/g` == '2' ] || error "target nlink"
 }
 run_test 31i "cross directory link under parent==============="
 
-
 test_31j() {
-        test_mkdir $DIR/d31j
-        test_mkdir $DIR/d31j/dir1
-        ln $DIR/d31j/dir1 $DIR/d31j/dir2 && error "ln for dir"
-        link $DIR/d31j/dir1 $DIR/d31j/dir3 && error "link for dir"
-        mlink $DIR/d31j/dir1 $DIR/d31j/dir4 && error "mlink for dir"
-        mlink $DIR/d31j/dir1 $DIR/d31j/dir1 && error "mlink to the same dir"
+	test_mkdir -c1 -p $DIR/$tdir
+	test_mkdir -c1 -p $DIR/$tdir/dir1
+	ln $DIR/$tdir/dir1 $DIR/$tdir/dir2 && error "ln for dir"
+	link $DIR/$tdir/dir1 $DIR/$tdir/dir3 && error "link for dir"
+	mlink $DIR/$tdir/dir1 $DIR/$tdir/dir4 && error "mlink for dir"
+	mlink $DIR/$tdir/dir1 $DIR/$tdir/dir1 && error "mlink to the same dir"
 	return 0
 }
 run_test 31j "link for directory==============="
 
-
 test_31k() {
-        test_mkdir $DIR/d31k
-        touch $DIR/d31k/s
-        touch $DIR/d31k/exist
-        mlink $DIR/d31k/s $DIR/d31k/t || error "mlink"
-        mlink $DIR/d31k/s $DIR/d31k/exist && error "mlink to exist file"
-        mlink $DIR/d31k/s $DIR/d31k/s && error "mlink to the same file"
-        mlink $DIR/d31k/s $DIR/d31k && error "mlink to parent dir"
-        mlink $DIR/d31k $DIR/d31k/s && error "mlink parent dir to target"
-        mlink $DIR/d31k/not-exist $DIR/d31k/foo && error "mlink non-existing to new"
-        mlink $DIR/d31k/not-exist $DIR/d31k/s && error "mlink non-existing to exist"
+        test_mkdir -c1 -p $DIR/$tdir
+        touch $DIR/$tdir/s
+        touch $DIR/$tdir/exist
+        mlink $DIR/$tdir/s $DIR/$tdir/t || error "mlink"
+        mlink $DIR/$tdir/s $DIR/$tdir/exist && error "mlink to exist file"
+        mlink $DIR/$tdir/s $DIR/$tdir/s && error "mlink to the same file"
+        mlink $DIR/$tdir/s $DIR/$tdir && error "mlink to parent dir"
+        mlink $DIR/$tdir $DIR/$tdir/s && error "mlink parent dir to target"
+        mlink $DIR/$tdir/not-exist $DIR/$tdir/foo && error "mlink non-existing to new"
+        mlink $DIR/$tdir/not-exist $DIR/$tdir/s && error "mlink non-existing to exist"
 	return 0
 }
 run_test 31k "link to file: the same, non-existing, dir==============="
 
 test_31m() {
-        test_mkdir $DIR/d31m
+        mkdir $DIR/d31m
         touch $DIR/d31m/s
-        test_mkdir $DIR/d31m2
+        mkdir $DIR/d31m2
         touch $DIR/d31m2/exist
         mlink $DIR/d31m/s $DIR/d31m2/t || error "mlink"
         mlink $DIR/d31m/s $DIR/d31m2/exist && error "mlink to exist file"
@@ -3990,6 +3988,16 @@ find_loop_dev() {
 	done
 }
 
+cleanup_54c() {
+	loopdev="$DIR/loop54c"
+
+	trap 0
+	$UMOUNT -d $tdir || rc=$?
+	losetup -d $loopdev || true
+	rm $loopdev
+	return $rc
+}
+
 test_54c() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
 	tfile="$DIR/f54c"
@@ -4000,17 +4008,16 @@ test_54c() {
 	[ -z "$LOOPNUM" ] && echo "couldn't find empty loop device" && return
 	mknod $loopdev b 7 $LOOPNUM
 	echo "make a loop file system with $tfile on $loopdev ($LOOPNUM)..."
-	dd if=/dev/zero of=$tfile bs=`page_size` seek=1024 count=1 > /dev/null
+	dd if=/dev/zero of=$tfile bs=$(get_page_size client) seek=1024 count=1 > /dev/null
 	losetup $loopdev $tfile || error "can't set up $loopdev for $tfile"
+	trap cleanup_54c EXIT
 	mkfs.ext2 $loopdev || error "mke2fs on $loopdev"
 	test_mkdir -p $tdir
 	mount -t ext2 $loopdev $tdir || error "error mounting $loopdev on $tdir"
 	dd if=/dev/zero of=$tdir/tmp bs=`page_size` count=30 || error "dd write"
 	df $tdir
 	dd if=$tdir/tmp of=/dev/zero bs=`page_size` count=30 || error "dd read"
-	$UMOUNT -d $tdir
-	losetup -d $loopdev
-	rm $loopdev
+	cleanup_54c
 }
 run_test 54c "block device works in lustre ====================="
 
@@ -4093,23 +4100,24 @@ run_test 56a "check $GETSTRIPE"
 NUMFILES=3
 NUMDIRS=3
 setup_56() {
-        local LOCAL_NUMFILES="$1"
-        local LOCAL_NUMDIRS="$2"
-        local MKDIR_PARAMS="$3"
-
-        if [ ! -d "$TDIR" ] ; then
-                test_mkdir -p $TDIR
-                [ "$MKDIR_PARAMS" ] && $SETSTRIPE $MKDIR_PARAMS $TDIR
-                for i in `seq 1 $LOCAL_NUMFILES` ; do
-                        touch $TDIR/file$i
-                done
-                for i in `seq 1 $LOCAL_NUMDIRS` ; do
-                        test_mkdir $TDIR/dir$i
-                        for j in `seq 1 $LOCAL_NUMFILES` ; do
-                                touch $TDIR/dir$i/file$j
-                        done
-                done
-        fi
+	local LOCAL_NUMFILES="$1"
+	local LOCAL_NUMDIRS="$2"
+	local MKDIR_PARAMS="$3"
+	local DIR_STRIPE_PARAMS="$4"
+
+	if [ ! -d "$TDIR" ] ; then
+		test_mkdir -p $DIR_STRIPE_PARAMS $TDIR
+		[ "$MKDIR_PARAMS" ] && $SETSTRIPE $MKDIR_PARAMS $TDIR
+		for i in `seq 1 $LOCAL_NUMFILES` ; do
+			touch $TDIR/file$i
+		done
+		for i in `seq 1 $LOCAL_NUMDIRS` ; do
+			test_mkdir $DIR_STRIPE_PARAMS $TDIR/dir$i
+			for j in `seq 1 $LOCAL_NUMFILES` ; do
+				touch $TDIR/dir$i/file$j
+			done
+		done
+	fi
 }
 
 setup_56_special() {
@@ -4237,7 +4245,6 @@ run_test 56n "check lfs find -type l ============================="
 test_56o() {
 	TDIR=$DIR/${tdir}o
 	setup_56 $NUMFILES $NUMDIRS
-
 	utime $TDIR/file1 > /dev/null || error "utime (1)"
 	utime $TDIR/file2 > /dev/null || error "utime (2)"
 	utime $TDIR/dir1 > /dev/null || error "utime (3)"
@@ -4528,7 +4535,7 @@ test_56w() {
 	TDIR=$DIR/${tdir}w
 
     rm -rf $TDIR || error "remove $TDIR failed"
-    setup_56 $NUMFILES $NUMDIRS "-c $OSTCOUNT"
+    setup_56 $NUMFILES $NUMDIRS "-c $OSTCOUNT" "-c1"
 
     local stripe_size
     stripe_size=$($GETSTRIPE -S -d $TDIR) ||
@@ -7368,7 +7375,7 @@ run_test 119d "The DIO path should try to send a new rpc once one is completed"
 
 test_120a() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-        test_mkdir -p $DIR/$tdir
+	test_mkdir -p $DIR/$tdir
         [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
                skip "no early lock cancel on server" && return 0
 
@@ -7378,12 +7385,16 @@ test_120a() {
 	# asynchronous object destroy at MDT could cause bl ast to client
 	cancel_lru_locks osc
 
-        stat $DIR/$tdir > /dev/null
-        can1=`lctl get_param -n ldlm.services.ldlm_canceld.stats | awk '/ldlm_cancel/ {print $2}'`
-        blk1=`lctl get_param -n ldlm.services.ldlm_cbd.stats | awk '/ldlm_bl_callback/ {print $2}'`
-        test_mkdir $DIR/$tdir/d1
-        can2=`lctl get_param -n ldlm.services.ldlm_canceld.stats | awk '/ldlm_cancel/ {print $2}'`
-        blk2=`lctl get_param -n ldlm.services.ldlm_cbd.stats | awk '/ldlm_bl_callback/ {print $2}'`
+	stat $DIR/$tdir > /dev/null
+	can1=$(lctl get_param -n ldlm.services.ldlm_canceld.stats |
+					awk '/ldlm_cancel/ {print $2}')
+	blk1=$(lctl get_param -n ldlm.services.ldlm_cbd.stats |
+					awk '/ldlm_bl_callback/ {print $2}')
+	test_mkdir -c1 $DIR/$tdir/d1
+        can2=$(lctl get_param -n ldlm.services.ldlm_canceld.stats |
+					awk '/ldlm_cancel/ {print $2}')
+	blk2=$(lctl get_param -n ldlm.services.ldlm_cbd.stats |
+					awk '/ldlm_bl_callback/ {print $2}')
         [ $can1 -eq $can2 ] || error $((can2-can1)) "cancel RPC occured."
         [ $blk1 -eq $blk2 ] || error $((blk2-blk1)) "blocking RPC occured."
         lru_resize_enable mdc
@@ -7393,8 +7404,8 @@ run_test 120a "Early Lock Cancel: mkdir test"
 
 test_120b() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-        test_mkdir -p $DIR/$tdir
-        [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
+        test_mkdir $DIR/$tdir
+        [ -z "$(lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel)" ] && \
                skip "no early lock cancel on server" && return 0
         lru_resize_disable mdc
         lru_resize_disable osc
@@ -7414,13 +7425,13 @@ run_test 120b "Early Lock Cancel: create test"
 
 test_120c() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-        test_mkdir -p $DIR/$tdir
-        [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
-               skip "no early lock cancel on server" && return 0
+	test_mkdir -c1 $DIR/$tdir
+	[ -z "$(lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel)" ] && \
+	       skip "no early lock cancel on server" && return 0
         lru_resize_disable mdc
         lru_resize_disable osc
-	test_mkdir -p $DIR/$tdir/d1
-	test_mkdir -p $DIR/$tdir/d2
+	test_mkdir -p -c1 $DIR/$tdir/d1
+	test_mkdir -p -c1 $DIR/$tdir/d2
         touch $DIR/$tdir/d1/f1
         cancel_lru_locks mdc
         stat $DIR/$tdir/d1 $DIR/$tdir/d2 $DIR/$tdir/d1/f1 > /dev/null
@@ -7438,9 +7449,9 @@ run_test 120c "Early Lock Cancel: link test"
 
 test_120d() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-        test_mkdir -p $DIR/$tdir
-        [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
-               skip "no early lock cancel on server" && return 0
+	test_mkdir -p -c1 $DIR/$tdir
+	[ -z "$(lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel)" ] && \
+	       skip "no early lock cancel on server" && return 0
         lru_resize_disable mdc
         lru_resize_disable osc
         touch $DIR/$tdir
@@ -7460,7 +7471,7 @@ run_test 120d "Early Lock Cancel: setattr test"
 
 test_120e() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-        test_mkdir -p $DIR/$tdir
+        test_mkdir -p -c1 $DIR/$tdir
         [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
                skip "no early lock cancel on server" && return 0
         lru_resize_disable mdc
@@ -7490,11 +7501,11 @@ test_120f() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
         [ -z "`lctl get_param -n mdc.*.connect_flags | grep early_lock_cancel`" ] && \
                skip "no early lock cancel on server" && return 0
-        test_mkdir -p $DIR/$tdir
+        test_mkdir -p -c1 $DIR/$tdir
         lru_resize_disable mdc
         lru_resize_disable osc
-	test_mkdir -p $DIR/$tdir/d1
-	test_mkdir -p $DIR/$tdir/d2
+	test_mkdir -p -c1 $DIR/$tdir/d1
+	test_mkdir -p -c1 $DIR/$tdir/d2
         dd if=/dev/zero of=$DIR/$tdir/d1/f1 count=1
         dd if=/dev/zero of=$DIR/$tdir/d2/f2 count=1
         cancel_lru_locks mdc
@@ -7981,7 +7992,6 @@ test_129() {
 		return
 	fi
 	remote_mds_nodsh && skip "remote MDS with nodsh" && return
-
 	ENOSPC=28
 	EFBIG=27
 
@@ -7995,6 +8005,9 @@ test_129() {
 	set_dir_limits $MAX
 	local I=$(stat -c%s "$DIR/$tdir")
 	local J=0
+	local STRIPE_COUNT=1
+	[ $MDSCOUNT -ge 2 ] && STRIPE_COUNT=$($LFS getdirstripe -c $DIR/$tdir)
+	MAX=$((MAX*STRIPE_COUNT))
 	while [ ! $I -gt $MAX ]; do
 		$MULTIOP $DIR/$tdir/$J Oc
 		rc=$?
@@ -8027,7 +8040,7 @@ test_129() {
 	done
 
 	set_dir_limits 0
-	error "exceeded dir size limit $MAX x $MDSCOUNT $MAX : $I bytes"
+	error "exceeded dir size limit $MAX($MDSCOUNT) : $I bytes"
 }
 run_test 129 "test directory size limit ========================"
 
@@ -9017,8 +9030,10 @@ dot_lustre_fid_permission_check() {
 	rm -f $test_dir/$tfile.1
 	echo "truncate fid $fid"
 	$TRUNCATE $ffid 777 || error "truncate $ffid failed."
-	echo "link fid $fid"
-	ln -f $ffid $test_dir/tfile.lnk || error "link $ffid failed."
+	if [ $MDSCOUNT -lt 2 ]; then #FIXME when cross-MDT hard link is working
+		echo "link fid $fid"
+		ln -f $ffid $test_dir/tfile.lnk || error "link $ffid failed."
+	fi
 	if [ -n $(lctl get_param -n mdc.*-mdc-*.connect_flags | grep acl) ]; then
 		echo "setfacl fid $fid"
 		setfacl -R -m u:bin:rwx $ffid || error "setfacl $ffid failed."
@@ -9169,7 +9184,7 @@ test_154b() {
 	local rc=0
 
 	mkdir -p $DIR/$tdir
-	$LFS mkdir -i $MDTIDX $remote_dir ||
+	$LFS mkdir -i $MDTIDX -c $MDSCOUNT $remote_dir ||
 		error "create remote directory failed"
 
 	cp /etc/hosts $remote_dir/$tfile
@@ -9672,14 +9687,14 @@ run_test 160b "Verify that very long rename doesn't crash in changelog"
 
 test_161a() {
 	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
-    test_mkdir -p $DIR/$tdir
-    cp /etc/hosts $DIR/$tdir/$tfile
-    test_mkdir $DIR/$tdir/foo1
-    test_mkdir $DIR/$tdir/foo2
-    ln $DIR/$tdir/$tfile $DIR/$tdir/foo1/sofia
-    ln $DIR/$tdir/$tfile $DIR/$tdir/foo2/zachary
-    ln $DIR/$tdir/$tfile $DIR/$tdir/foo1/luna
-    ln $DIR/$tdir/$tfile $DIR/$tdir/foo2/thor
+	test_mkdir -p -c1 $DIR/$tdir
+	cp /etc/hosts $DIR/$tdir/$tfile
+	test_mkdir -c1 $DIR/$tdir/foo1
+	test_mkdir -c1 $DIR/$tdir/foo2
+	ln $DIR/$tdir/$tfile $DIR/$tdir/foo1/sofia
+	ln $DIR/$tdir/$tfile $DIR/$tdir/foo2/zachary
+	ln $DIR/$tdir/$tfile $DIR/$tdir/foo1/luna
+	ln $DIR/$tdir/$tfile $DIR/$tdir/foo2/thor
 	local FID=$($LFS path2fid $DIR/$tdir/$tfile | tr -d '[]')
 	if [ "$($LFS fid2path $DIR $FID | wc -l)" != "5" ]; then
 		$LFS fid2path $DIR $FID
@@ -10207,7 +10222,7 @@ test_184a() {
 	check_swap_layouts_support && return 0
 
 	dir0=$DIR/$tdir/$testnum
-	test_mkdir -p $dir0 || error "creating dir $dir0"
+	test_mkdir -p -c1 $dir0 || error "creating dir $dir0"
 	ref1=/etc/passwd
 	ref2=/etc/group
 	file1=$dir0/f1
@@ -11903,7 +11918,7 @@ run_test 235 "LU-1715: flock deadlock detection does not work properly"
 #LU-2935
 test_236() {
 	check_swap_layouts_support && return 0
-	test_mkdir -p $DIR/$tdir || error "mkdir $tdir failed"
+	test_mkdir -p -c1 $DIR/$tdir || error "mkdir $tdir failed"
 
 	local ref1=/etc/passwd
 	local ref2=/etc/group
@@ -11940,6 +11955,236 @@ test_237() {
 }
 run_test 237 "Verify name_to_handle_at/open_by_handle_at syscalls"
 
+test_striped_dir() {
+	local mdt_index=$1
+	local stripe_count
+	local stripe_index
+
+	mkdir -p $DIR/$tdir
+	$LFS setdirstripe -i $mdt_index -c 2 -t all_char $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+
+	stripe_count=$($LFS getdirstripe -c $DIR/$tdir/striped_dir)
+	if [ "$stripe_count" != "2" ]; then
+		error "stripe_count is $stripe_count, expect 2"
+	fi
+
+	stripe_index=$($LFS getdirstripe -i $DIR/$tdir/striped_dir)
+	if [ "$stripe_index" != "$mdt_index" ]; then
+		error "stripe_index is $stripe_index, expect $mdt_index"
+	fi
+
+	[ $(stat -c%h $DIR/$tdir/striped_dir) == '2' ] ||
+		error "nlink error after create striped dir"
+
+	mkdir $DIR/$tdir/striped_dir/a
+	mkdir $DIR/$tdir/striped_dir/b
+
+	stat $DIR/$tdir/striped_dir/a ||
+		error "create dir under striped dir failed"
+	stat $DIR/$tdir/striped_dir/b ||
+		error "create dir under striped dir failed"
+
+	[ $(stat -c%h $DIR/$tdir/striped_dir) == '4' ] ||
+		error "nlink error after mkdir"
+
+	rmdir $DIR/$tdir/striped_dir/a
+	[ $(stat -c%h $DIR/$tdir/striped_dir) == '3' ] ||
+		error "nlink error after rmdir"
+
+	rmdir $DIR/$tdir/striped_dir/b
+	[ $(stat -c%h $DIR/$tdir/striped_dir) == '2' ] ||
+		error "nlink error after rmdir"
+
+	rmdir $DIR/$tdir/striped_dir ||
+		error "rmdir striped dir error"
+	true
+}
+
+test_300a() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+
+	test_striped_dir 0 || error "failed on striped dir on MDT0"
+	test_striped_dir 1 || error "failed on striped dir on MDT0"
+}
+run_test 300a "basic striped dir sanity test"
+
+test_300b() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+	local i
+	local mtime1
+	local mtime2
+	local mtime3
+
+	test_mkdir $DIR/$tdir || error "mkdir fail"
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+	for ((i=0; i<10; i++)); do
+		mtime1=$(stat -c %Y $DIR/$tdir/striped_dir)
+		sleep 1
+		touch $DIR/$tdir/striped_dir/file_$i ||
+					error "touch error $i"
+		mtime2=$(stat -c %Y $DIR/$tdir/striped_dir)
+		[ $mtime1 -eq $mtime2 ] &&
+			error "mtime not change after create"
+		sleep 1
+		rm -f $DIR/$tdir/striped_dir/file_$i ||
+					error "unlink error $i"
+		mtime3=$(stat -c %Y $DIR/$tdir/striped_dir)
+		[ $mtime2 -eq $mtime3 ] &&
+			error "mtime did not change after unlink"
+	done
+	true
+}
+run_test 300b "check ctime/mtime for striped dir"
+
+test_300c() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+	local file_count
+
+	mkdir -p $DIR/$tdir
+	$LFS setdirstripe -i 0 -c 2 $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+
+	chown $RUNAS_ID:$RUNAS_GID $DIR/$tdir/striped_dir ||
+		error "chown striped dir failed"
+
+	$RUNAS createmany -o $DIR/$tdir/striped_dir/f 5000 ||
+		error "create 5k files failed"
+
+	file_count=$(ls $DIR/$tdir/striped_dir | wc -l)
+
+	[ "$file_count" = 5000 ] || error "file count $file_count != 5000"
+
+	rm -rf $DIR/$tdir
+}
+run_test 300c "chown && check ls under striped directory"
+
+test_300d() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+	local stripe_count
+	local file
+
+	mkdir -p $DIR/$tdir
+	$SETSTRIPE -c 2 $DIR/$tdir
+
+	#local striped directory
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+	createmany -o $DIR/$tdir/striped_dir/f 10 ||
+		error "create 10 files failed"
+
+	#remote striped directory
+	$LFS setdirstripe -i 1 -c 2 $DIR/$tdir/remote_striped_dir ||
+		error "set striped dir error"
+	createmany -o $DIR/$tdir/remote_striped_dir/f 10 ||
+		error "create 10 files failed"
+
+	for file in $(find $DIR/$tdir); do
+		stripe_count=$($GETSTRIPE -c $file)
+		[ $stripe_count -eq 2 ] ||
+			error "wrong stripe $stripe_count for $file"
+	done
+
+	rm -rf $DIR/$tdir
+}
+run_test 300d "check default stripe under striped directory"
+
+test_300e() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+	local stripe_count
+	local file
+
+	mkdir -p $DIR/$tdir
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+
+	touch $DIR/$tdir/striped_dir/a
+	touch $DIR/$tdir/striped_dir/b
+	touch $DIR/$tdir/striped_dir/c
+
+	mkdir $DIR/$tdir/striped_dir/dir_a
+	mkdir $DIR/$tdir/striped_dir/dir_b
+	mkdir $DIR/$tdir/striped_dir/dir_c
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir/stp_a ||
+		error "set striped dir under striped dir error"
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir/stp_b ||
+		error "set striped dir under striped dir error"
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir/stp_c ||
+		error "set striped dir under striped dir error"
+
+	mrename $DIR/$tdir/striped_dir/a $DIR/$tdir/striped_dir/b &&
+		error "rename file under striped dir should fail"
+
+	mrename $DIR/$tdir/striped_dir/dir_a $DIR/$tdir/striped_dir/dir_b &&
+		error "rename dir under striped dir should fail"
+
+	mrename $DIR/$tdir/striped_dir/stp_a $DIR/$tdir/striped_dir/stp_b &&
+		error "rename dir under different stripes should fail"
+
+	mrename $DIR/$tdir/striped_dir/a $DIR/$tdir/striped_dir/c ||
+		error "rename file under striped dir should succeed"
+
+	mrename $DIR/$tdir/striped_dir/dir_a $DIR/$tdir/striped_dir/dir_c ||
+		error "rename dir under striped dir should succeed"
+
+	rm -rf $DIR/$tdir
+}
+run_test 300e "check rename under striped directory"
+
+test_300f() {
+	[ $PARALLEL == "yes" ] && skip "skip parallel run" && return
+	[ $MDSCOUNT -lt 2 ] && skip "needs >= 2 MDTs" && return
+	local stripe_count
+	local file
+
+	rm -rf $DIR/$tdir
+	mkdir -p $DIR/$tdir
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir ||
+		error "set striped dir error"
+
+	$LFS setdirstripe -i 0 -c 2 -t all_char $DIR/$tdir/striped_dir1 ||
+		error "set striped dir error"
+
+	touch $DIR/$tdir/striped_dir/a
+	mkdir $DIR/$tdir/striped_dir/dir_a
+	$LFS setdirstripe -i 0 -c 2 $DIR/$tdir/striped_dir/stp_a ||
+		error "create striped dir under striped dir fails"
+
+	touch $DIR/$tdir/striped_dir1/b
+	mkdir $DIR/$tdir/striped_dir1/dir_b
+	$LFS setdirstripe -i 0 -c 2 $DIR/$tdir/striped_dir/stp_b ||
+		error "create striped dir under striped dir fails"
+
+	mrename $DIR/$tdir/striped_dir/a $DIR/$tdir/striped_dir1/b &&
+		error "rename file under different striped dir should fail"
+
+	mrename $DIR/$tdir/striped_dir/dir_a $DIR/$tdir/striped_dir1/dir_b &&
+		error "rename dir under different striped dir should fail"
+
+	mrename $DIR/$tdir/striped_dir/stp_a $DIR/$tdir/striped_dir1/stp_b &&
+		error "rename striped dir under diff striped dir should fail"
+
+	mrename $DIR/$tdir/striped_dir/a $DIR/$tdir/striped_dir1/a ||
+		error "rename file under diff striped dirs fails"
+
+	mrename $DIR/$tdir/striped_dir/dir_a $DIR/$tdir/striped_dir1/dir_a ||
+		error "rename dir under diff striped dirs fails"
+
+	rm -rf $DIR/$tdir
+}
+run_test 300f "check rename cross striped directory"
+
 #
 # tests that do cleanup/setup should be run at the end
 #
diff --git a/lustre/tests/sanityn.sh b/lustre/tests/sanityn.sh
index b6222a7d..09524f8 100644
--- a/lustre/tests/sanityn.sh
+++ b/lustre/tests/sanityn.sh
@@ -2549,6 +2549,7 @@ run_test 75 "osc: upcall after unuse lock==================="
 test_76() { #LU-946
 	[[ $(lustre_version_code $SINGLEMDS) -lt $(version_code 2.5.53) ]] &&
 		skip "Need MDS version at least 2.5.53" && return
+	[ $MDSCOUNT -ge 2 ] && skip "skip now for LU-4573" && return #LU-4573
 
 	remote_mds_nodsh && skip "remote MDS with nodsh" && return
 	local fcount=2048
diff --git a/lustre/tests/test-framework.sh b/lustre/tests/test-framework.sh
index 0292bc2..8744cc5 100644
--- a/lustre/tests/test-framework.sh
+++ b/lustre/tests/test-framework.sh
@@ -6702,24 +6702,41 @@ test_mkdir() {
 	local parent
 	local child
 	local path
+	local p_option
+	local option2
+	local stripe_count=2
 	local rc=0
 
 	case $# in
 		1) path=$1;;
 		2) option=$1
 		   path=$2;;
+		3) option=$1
+		   option2=$2
+		   path=$3;;
 		*) error "Only creating single directory is supported";;
 	esac
 
 	child=$(basename $path)
 	parent=$(dirname $path)
 
-	if [ "$option" == "-p" -a -d $parent/$child ]; then
-		return $rc
+	if [ "$option" == "-p" -o "$option2" == "-p" ]; then
+		if [ -d $parent/$child ]; then
+			return $rc
+		fi
+		p_option="-p"
+	fi
+
+	if [ "${option:0:2}" == "-c" ]; then
+		stripe_count=$(echo $option | sed 's/^-c//')
+	fi
+
+	if [ "${option2:0:2}" == "-c" ]; then
+		stripe_count=$(echo $option2 | sed 's/^-c//')
 	fi
 
 	if [ ! -d ${parent} ]; then
-		if [ "$option" == "-p" ]; then
+		if [ "$p_option" == "-p" ]; then
 			mkdir -p ${parent}
 		else
 			return 1
@@ -6727,18 +6744,14 @@ test_mkdir() {
 	fi
 
 	if [ $MDSCOUNT -le 1 ]; then
-		mkdir $option $parent/$child || rc=$?
+		mkdir $p_option $parent/$child || rc=$?
 	else
 		local mdt_idx=$($LFS getstripe -M $parent)
 		local test_num=$(echo $testnum | sed -e 's/[^0-9]*//g')
 
-		if [ "$mdt_idx" -ne 0 ]; then
-			mkdir $option $parent/$child || rc=$?
-		else
-			mdt_idx=$((test_num % MDSCOUNT))
-			echo "mkdir $mdt_idx for $parent/$child"
-			$LFS setdirstripe -i $mdt_idx $parent/$child || rc=$?
-		fi
+		mdt_idx=$((test_num % MDSCOUNT))
+		echo "striped dir -i$mdt_idx -c$stripe_count $path"
+		$LFS setdirstripe -i$mdt_idx -c$stripe_count $path || rc=$?
 	fi
 	return $rc
 }