X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdt%2Fmdt_handler.c;h=acae12269298b0e33684ec10013c81c71f94394f;hp=9f84d086ae76d51da2a92831a1ba225abf9f277d;hb=109cdb83d77844e00ebc6d5df6bf73845cf9d45e;hpb=d7756663bc659044b3a63a3236d6d5fec264add2

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 9f84d08..acae122 100644
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -27,7 +27,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2010, 2013, Intel Corporation.
+ * Copyright (c) 2010, 2015, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -49,26 +49,30 @@
 #define DEBUG_SUBSYSTEM S_MDS
 
 #include <linux/module.h>
-/*
- * struct OBD_{ALLOC,FREE}*()
- */
-#include <obd_support.h>
+
+#include <dt_object.h>
+#include <lustre_acl.h>
+#include <lustre_export.h>
 #include <lustre_ioctl.h>
-/* struct ptlrpc_request */
+#include <lustre_lfsck.h>
+#include <lustre_log.h>
 #include <lustre_net.h>
-/* struct obd_export */
-#include <lustre_export.h>
-/* struct obd_device */
-#include <obd.h>
-/* lu2dt_dev() */
-#include <dt_object.h>
+#include <lustre_nodemap.h>
 #include <lustre_mds.h>
-#include <lustre_log.h>
-#include "mdt_internal.h"
-#include <lustre_acl.h>
 #include <lustre_param.h>
 #include <lustre_quota.h>
-#include <lustre_lfsck.h>
+#include <lustre_swab.h>
+#include <obd.h>
+#include <obd_support.h>
+
+#include <llog_swab.h>
+
+#include "mdt_internal.h"
+
+
+static unsigned int max_mod_rpcs_per_client = 8;
+module_param(max_mod_rpcs_per_client, uint, 0644);
+MODULE_PARM_DESC(max_mod_rpcs_per_client, "maximum number of modify RPCs in flight allowed per client");
 
 mdl_mode_t mdt_mdl_lock_modes[] = {
         [LCK_MINMODE] = MDL_MINMODE,
@@ -81,15 +85,15 @@ mdl_mode_t mdt_mdl_lock_modes[] = {
         [LCK_GROUP]   = MDL_GROUP
 };
 
-ldlm_mode_t mdt_dlm_lock_modes[] = {
-        [MDL_MINMODE] = LCK_MINMODE,
-        [MDL_EX]      = LCK_EX,
-        [MDL_PW]      = LCK_PW,
-        [MDL_PR]      = LCK_PR,
-        [MDL_CW]      = LCK_CW,
-        [MDL_CR]      = LCK_CR,
-        [MDL_NL]      = LCK_NL,
-        [MDL_GROUP]   = LCK_GROUP
+enum ldlm_mode mdt_dlm_lock_modes[] = {
+	[MDL_MINMODE]	= LCK_MINMODE,
+	[MDL_EX]	= LCK_EX,
+	[MDL_PW]	= LCK_PW,
+	[MDL_PR]	= LCK_PR,
+	[MDL_CW]	= LCK_CW,
+	[MDL_CR]	= LCK_CR,
+	[MDL_NL]	= LCK_NL,
+	[MDL_GROUP]	= LCK_GROUP
 };
 
 static struct mdt_device *mdt_dev(struct lu_device *d);
@@ -127,47 +131,48 @@ static struct lu_kmem_descr mdt_caches[] = {
 	}
 };
 
-int mdt_get_disposition(struct ldlm_reply *rep, int flag)
+__u64 mdt_get_disposition(struct ldlm_reply *rep, __u64 op_flag)
 {
-        if (!rep)
-                return 0;
-        return (rep->lock_policy_res1 & flag);
+	if (!rep)
+		return 0;
+	return rep->lock_policy_res1 & op_flag;
 }
 
 void mdt_clear_disposition(struct mdt_thread_info *info,
-			   struct ldlm_reply *rep, int flag)
+			   struct ldlm_reply *rep, __u64 op_flag)
 {
 	if (info) {
-		info->mti_opdata &= ~flag;
-		tgt_opdata_clear(info->mti_env, flag);
+		info->mti_opdata &= ~op_flag;
+		tgt_opdata_clear(info->mti_env, op_flag);
 	}
 	if (rep)
-		rep->lock_policy_res1 &= ~flag;
+		rep->lock_policy_res1 &= ~op_flag;
 }
 
 void mdt_set_disposition(struct mdt_thread_info *info,
-			 struct ldlm_reply *rep, int flag)
+			 struct ldlm_reply *rep, __u64 op_flag)
 {
 	if (info) {
-		info->mti_opdata |= flag;
-		tgt_opdata_set(info->mti_env, flag);
+		info->mti_opdata |= op_flag;
+		tgt_opdata_set(info->mti_env, op_flag);
 	}
 	if (rep)
-		rep->lock_policy_res1 |= flag;
+		rep->lock_policy_res1 |= op_flag;
 }
 
-void mdt_lock_reg_init(struct mdt_lock_handle *lh, ldlm_mode_t lm)
+void mdt_lock_reg_init(struct mdt_lock_handle *lh, enum ldlm_mode lm)
 {
-        lh->mlh_pdo_hash = 0;
-        lh->mlh_reg_mode = lm;
+	lh->mlh_pdo_hash = 0;
+	lh->mlh_reg_mode = lm;
 	lh->mlh_rreg_mode = lm;
-        lh->mlh_type = MDT_REG_LOCK;
+	lh->mlh_type = MDT_REG_LOCK;
 }
 
-void mdt_lock_pdo_init(struct mdt_lock_handle *lh, ldlm_mode_t lock_mode,
+void mdt_lock_pdo_init(struct mdt_lock_handle *lh, enum ldlm_mode lock_mode,
 		       const struct lu_name *lname)
 {
 	lh->mlh_reg_mode = lock_mode;
+	lh->mlh_pdo_mode = LCK_MINMODE;
 	lh->mlh_rreg_mode = lock_mode;
 	lh->mlh_type = MDT_PDO_LOCK;
 
@@ -263,11 +268,95 @@ static void mdt_lock_pdo_mode(struct mdt_thread_info *info, struct mdt_object *o
         EXIT;
 }
 
-static int mdt_getstatus(struct tgt_session_info *tsi)
+static int mdt_lookup_fileset(struct mdt_thread_info *info, const char *fileset,
+			      struct lu_fid *fid)
+{
+	struct mdt_device *mdt = info->mti_mdt;
+	struct lu_name *lname = &info->mti_name;
+	char *name = NULL;
+	struct mdt_object *parent;
+	u32 mode;
+	int rc = 0;
+
+	LASSERT(!info->mti_cross_ref);
+
+	OBD_ALLOC(name, NAME_MAX + 1);
+	if (name == NULL)
+		return -ENOMEM;
+	lname->ln_name = name;
+
+	/*
+	 * We may want to allow this to mount a completely separate
+	 * fileset from the MDT in the future, but keeping it to
+	 * ROOT/ only for now avoid potential security issues.
+	 */
+	*fid = mdt->mdt_md_root_fid;
+
+	while (rc == 0 && fileset != NULL && *fileset != '\0') {
+		const char *s1 = fileset;
+		const char *s2;
+
+		while (*++s1 == '/')
+			;
+		s2 = s1;
+		while (*s2 != '/' && *s2 != '\0')
+			s2++;
+
+		if (s2 == s1)
+			break;
+
+		fileset = s2;
+
+		lname->ln_namelen = s2 - s1;
+		if (lname->ln_namelen > NAME_MAX) {
+			rc = -EINVAL;
+			break;
+		}
+
+		/* reject .. as a path component */
+		if (lname->ln_namelen == 2 &&
+		    strncmp(s1, "..", 2) == 0) {
+			rc = -EINVAL;
+			break;
+		}
+
+		strncpy(name, s1, lname->ln_namelen);
+		name[lname->ln_namelen] = '\0';
+
+		parent = mdt_object_find(info->mti_env, mdt, fid);
+		if (IS_ERR(parent)) {
+			rc = PTR_ERR(parent);
+			break;
+		}
+		/* Only got the fid of this obj by name */
+		fid_zero(fid);
+		rc = mdo_lookup(info->mti_env, mdt_object_child(parent), lname,
+				fid, &info->mti_spec);
+		mdt_object_put(info->mti_env, parent);
+	}
+	if (!rc) {
+		parent = mdt_object_find(info->mti_env, mdt, fid);
+		if (IS_ERR(parent))
+			rc = PTR_ERR(parent);
+		else {
+			mode = lu_object_attr(&parent->mot_obj);
+			mdt_object_put(info->mti_env, parent);
+			if (!S_ISDIR(mode))
+				rc = -ENOTDIR;
+		}
+	}
+
+	OBD_FREE(name, NAME_MAX + 1);
+
+	return rc;
+}
+
+static int mdt_get_root(struct tgt_session_info *tsi)
 {
 	struct mdt_thread_info	*info = tsi2mdt_info(tsi);
 	struct mdt_device	*mdt = info->mti_mdt;
 	struct mdt_body		*repbody;
+	char			*fileset;
 	int			 rc;
 
 	ENTRY;
@@ -276,31 +365,23 @@ static int mdt_getstatus(struct tgt_session_info *tsi)
 	if (rc)
 		GOTO(out, rc = err_serious(rc));
 
-	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETSTATUS_PACK))
+	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GET_ROOT_PACK))
 		GOTO(out, rc = err_serious(-ENOMEM));
 
 	repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
-	repbody->fid1 = mdt->mdt_md_root_fid;
-	repbody->valid |= OBD_MD_FLID;
+	if (req_capsule_get_size(info->mti_pill, &RMF_NAME, RCL_CLIENT) > 0) {
+		fileset = req_capsule_client_get(info->mti_pill, &RMF_NAME);
+		if (fileset == NULL)
+			GOTO(out, rc = err_serious(-EFAULT));
 
-	if (tsi->tsi_tgt->lut_mds_capa &&
-	    exp_connect_flags(info->mti_exp) & OBD_CONNECT_MDS_CAPA) {
-		struct mdt_object	*root;
-		struct lustre_capa	*capa;
+		rc = mdt_lookup_fileset(info, fileset, &repbody->mbo_fid1);
+		if (rc < 0)
+			GOTO(out, rc = err_serious(rc));
+	} else {
+		repbody->mbo_fid1 = mdt->mdt_md_root_fid;
+	}
+	repbody->mbo_valid |= OBD_MD_FLID;
 
-		root = mdt_object_find(info->mti_env, mdt, &repbody->fid1);
-		if (IS_ERR(root))
-			GOTO(out, rc = PTR_ERR(root));
-
-                capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1);
-                LASSERT(capa);
-                capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
-                rc = mo_capa_get(info->mti_env, mdt_object_child(root), capa,
-                                 0);
-                mdt_object_put(info->mti_env, root);
-                if (rc == 0)
-                        repbody->valid |= OBD_MD_FLMDSCAPA;
-        }
 	EXIT;
 out:
 	mdt_thread_info_fini(info);
@@ -361,100 +442,159 @@ out:
 	RETURN(rc);
 }
 
-/**
- * Pack SOM attributes into the reply.
- * Call under a DLM UPDATE lock.
+#ifdef CONFIG_FS_POSIX_ACL
+/*
+ * Pack ACL data into the reply. UIDs/GIDs are mapped and filtered by nodemap.
+ *
+ * \param	info	thread info object
+ * \param	repbody	reply to pack ACLs into
+ * \param	o	mdt object of file to examine
+ * \param	nodemap	nodemap of client to reply to
+ * \retval	0	success
+ * \retval	-errno	error getting or parsing ACL from disk
  */
-static void mdt_pack_size2body(struct mdt_thread_info *info,
-                               struct mdt_object *mo)
-{
-        struct mdt_body *b;
-        struct md_attr *ma = &info->mti_attr;
-
-        LASSERT(ma->ma_attr.la_valid & LA_MODE);
-        b = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
-
-        /* Check if Size-on-MDS is supported, if this is a regular file,
-         * if SOM is enabled on the object and if SOM cache exists and valid.
-         * Otherwise do not pack Size-on-MDS attributes to the reply. */
-        if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM) ||
-            !S_ISREG(ma->ma_attr.la_mode) ||
-            !mdt_object_is_som_enabled(mo) ||
-            !(ma->ma_valid & MA_SOM))
-                return;
+int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody,
+		      struct mdt_object *o, struct lu_nodemap *nodemap)
+{
+	const struct lu_env	*env = info->mti_env;
+	struct md_object	*next = mdt_object_child(o);
+	struct lu_buf		*buf = &info->mti_buf;
+	int rc;
+
+	buf->lb_buf = req_capsule_server_get(info->mti_pill, &RMF_ACL);
+	buf->lb_len = req_capsule_get_size(info->mti_pill, &RMF_ACL,
+					   RCL_SERVER);
+	if (buf->lb_len == 0)
+		return 0;
 
-        b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
-        b->size = ma->ma_som->msd_size;
-        b->blocks = ma->ma_som->msd_blocks;
+	rc = mo_xattr_get(env, next, buf, XATTR_NAME_ACL_ACCESS);
+	if (rc < 0) {
+		if (rc == -ENODATA) {
+			repbody->mbo_aclsize = 0;
+			repbody->mbo_valid |= OBD_MD_FLACL;
+			rc = 0;
+		} else if (rc == -EOPNOTSUPP) {
+			rc = 0;
+		} else {
+			CERROR("%s: unable to read "DFID" ACL: rc = %d\n",
+			       mdt_obd_name(info->mti_mdt),
+			       PFID(mdt_object_fid(o)), rc);
+		}
+	} else {
+		rc = nodemap_map_acl(nodemap, buf->lb_buf,
+				     rc, NODEMAP_FS_TO_CLIENT);
+		/* if all ACLs mapped out, rc is still >= 0 */
+		if (rc < 0) {
+			CERROR("%s: nodemap_map_acl unable to parse "DFID
+			       " ACL: rc = %d\n", mdt_obd_name(info->mti_mdt),
+			       PFID(mdt_object_fid(o)), rc);
+		} else {
+			repbody->mbo_aclsize = rc;
+			repbody->mbo_valid |= OBD_MD_FLACL;
+			rc = 0;
+		}
+	}
+	return rc;
 }
+#endif
 
 void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
                         const struct lu_attr *attr, const struct lu_fid *fid)
 {
-        struct md_attr *ma = &info->mti_attr;
-
-        LASSERT(ma->ma_valid & MA_INODE);
-
-        b->atime      = attr->la_atime;
-        b->mtime      = attr->la_mtime;
-        b->ctime      = attr->la_ctime;
-        b->mode       = attr->la_mode;
-        b->size       = attr->la_size;
-        b->blocks     = attr->la_blocks;
-        b->uid        = attr->la_uid;
-        b->gid        = attr->la_gid;
-        b->flags      = attr->la_flags;
-        b->nlink      = attr->la_nlink;
-        b->rdev       = attr->la_rdev;
-
-        /*XXX should pack the reply body according to lu_valid*/
-        b->valid |= OBD_MD_FLCTIME | OBD_MD_FLUID   |
-                    OBD_MD_FLGID   | OBD_MD_FLTYPE  |
-                    OBD_MD_FLMODE  | OBD_MD_FLNLINK | OBD_MD_FLFLAGS |
-                    OBD_MD_FLATIME | OBD_MD_FLMTIME ;
-
-        if (!S_ISREG(attr->la_mode)) {
-                b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV;
+	struct md_attr		*ma = &info->mti_attr;
+	struct obd_export	*exp = info->mti_exp;
+	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
+
+	LASSERT(ma->ma_valid & MA_INODE);
+
+	if (attr->la_valid & LA_ATIME) {
+		b->mbo_atime = attr->la_atime;
+		b->mbo_valid |= OBD_MD_FLATIME;
+	}
+	if (attr->la_valid & LA_MTIME) {
+		b->mbo_mtime = attr->la_mtime;
+		b->mbo_valid |= OBD_MD_FLMTIME;
+	}
+	if (attr->la_valid & LA_CTIME) {
+		b->mbo_ctime = attr->la_ctime;
+		b->mbo_valid |= OBD_MD_FLCTIME;
+	}
+	if (attr->la_valid & LA_FLAGS) {
+		b->mbo_flags = attr->la_flags;
+		b->mbo_valid |= OBD_MD_FLFLAGS;
+	}
+	if (attr->la_valid & LA_NLINK) {
+		b->mbo_nlink = attr->la_nlink;
+		b->mbo_valid |= OBD_MD_FLNLINK;
+	}
+	if (attr->la_valid & LA_UID) {
+		b->mbo_uid = nodemap_map_id(nodemap, NODEMAP_UID,
+					    NODEMAP_FS_TO_CLIENT,
+					    attr->la_uid);
+		b->mbo_valid |= OBD_MD_FLUID;
+	}
+	if (attr->la_valid & LA_GID) {
+		b->mbo_gid = nodemap_map_id(nodemap, NODEMAP_GID,
+					    NODEMAP_FS_TO_CLIENT,
+					    attr->la_gid);
+		b->mbo_valid |= OBD_MD_FLGID;
+	}
+	b->mbo_mode = attr->la_mode;
+	if (attr->la_valid & LA_MODE)
+		b->mbo_valid |= OBD_MD_FLMODE;
+	if (attr->la_valid & LA_TYPE)
+		b->mbo_valid |= OBD_MD_FLTYPE;
+
+	if (fid != NULL) {
+		b->mbo_fid1 = *fid;
+		b->mbo_valid |= OBD_MD_FLID;
+		CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, valid="LPX64"\n",
+		       PFID(fid), b->mbo_nlink, b->mbo_mode, b->mbo_valid);
+	}
+
+	if (info != NULL)
+		mdt_body_reverse_idmap(info, b);
+
+	if (!(attr->la_valid & LA_TYPE))
+		return;
+
+	b->mbo_rdev   = attr->la_rdev;
+	b->mbo_size   = attr->la_size;
+	b->mbo_blocks = attr->la_blocks;
+
+	if (!S_ISREG(attr->la_mode)) {
+		b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS | OBD_MD_FLRDEV;
 	} else if (ma->ma_need & MA_LOV && !(ma->ma_valid & MA_LOV)) {
-                /* means no objects are allocated on osts. */
-                LASSERT(!(ma->ma_valid & MA_LOV));
-                /* just ignore blocks occupied by extend attributes on MDS */
-                b->blocks = 0;
-                /* if no object is allocated on osts, the size on mds is valid. b=22272 */
-                b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
+		/* means no objects are allocated on osts. */
+		LASSERT(!(ma->ma_valid & MA_LOV));
+		/* just ignore blocks occupied by extend attributes on MDS */
+		b->mbo_blocks = 0;
+		/* if no object is allocated on osts, the size on mds is valid.
+		 * b=22272 */
+		b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 	} else if ((ma->ma_valid & MA_LOV) && ma->ma_lmm != NULL &&
 		   ma->ma_lmm->lmm_pattern & LOV_PATTERN_F_RELEASED) {
 		/* A released file stores its size on MDS. */
 		/* But return 1 block for released file, unless tools like tar
 		 * will consider it fully sparse. (LU-3864)
 		 */
-		if (unlikely(b->size == 0))
-			b->blocks = 0;
+		if (unlikely(b->mbo_size == 0))
+			b->mbo_blocks = 0;
 		else
-			b->blocks = 1;
-		b->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
+			b->mbo_blocks = 1;
+		b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
 	}
 
-        if (fid) {
-                b->fid1 = *fid;
-                b->valid |= OBD_MD_FLID;
-                CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n",
-                                PFID(fid), b->nlink, b->mode, b->size);
-        }
-
-        if (info)
-                mdt_body_reverse_idmap(info, b);
-
-	if (fid != NULL && (b->valid & OBD_MD_FLSIZE))
-                CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
-                       PFID(fid), (unsigned long long)b->size);
+	if (fid != NULL && (b->mbo_valid & OBD_MD_FLSIZE))
+		CDEBUG(D_VFSTRACE, DFID": returning size %llu\n",
+		       PFID(fid), (unsigned long long)b->mbo_size);
 }
 
 static inline int mdt_body_has_lov(const struct lu_attr *la,
-                                   const struct mdt_body *body)
+				   const struct mdt_body *body)
 {
-        return ((S_ISREG(la->la_mode) && (body->valid & OBD_MD_FLEASIZE)) ||
-                (S_ISDIR(la->la_mode) && (body->valid & OBD_MD_FLDIREA )) );
+	return (S_ISREG(la->la_mode) && (body->mbo_valid & OBD_MD_FLEASIZE)) ||
+	       (S_ISDIR(la->la_mode) && (body->mbo_valid & OBD_MD_FLDIREA));
 }
 
 void mdt_client_compatibility(struct mdt_thread_info *info)
@@ -482,7 +622,8 @@ void mdt_client_compatibility(struct mdt_thread_info *info)
         EXIT;
 }
 
-int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, struct mdt_object *o)
+static int mdt_attr_get_eabuf_size(struct mdt_thread_info *info,
+				   struct mdt_object *o)
 {
 	const struct lu_env *env = info->mti_env;
 	int rc, rc2;
@@ -500,6 +641,12 @@ int mdt_attr_get_eabuf_size(struct mdt_thread_info *info, struct mdt_object *o)
 	if (S_ISDIR(lu_object_attr(&mdt_object_child(o)->mo_lu))) {
 		rc2 = mo_xattr_get(env, mdt_object_child(o), &LU_BUF_NULL,
 				   XATTR_NAME_LMV);
+
+		if (rc2 == -ENODATA)
+			rc2 = mo_xattr_get(env, mdt_object_child(o),
+					   &LU_BUF_NULL,
+					   XATTR_NAME_DEFAULT_LMV);
+
 		if ((rc2 < 0 && rc2 != -ENODATA) || (rc2 > rc))
 			rc = rc2;
 	}
@@ -508,8 +655,8 @@ out:
 	return rc;
 }
 
-static int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o,
-			     const char *name)
+int mdt_big_xattr_get(struct mdt_thread_info *info, struct mdt_object *o,
+		      const char *name)
 {
 	const struct lu_env *env = info->mti_env;
 	int rc;
@@ -572,10 +719,26 @@ int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o,
 
 	rc = mo_xattr_get(info->mti_env, next, buf, name);
 	if (rc > 0) {
+
+got:
 		if (strcmp(name, XATTR_NAME_LOV) == 0) {
-			ma->ma_lmm_size = rc;
-			ma->ma_valid |= MA_LOV;
+			if (info->mti_big_lmm_used)
+				ma->ma_lmm = info->mti_big_lmm;
+
+			/* NOT return LOV EA with hole to old client. */
+			if (unlikely(le32_to_cpu(ma->ma_lmm->lmm_pattern) &
+				     LOV_PATTERN_F_HOLE) &&
+			    !(exp_connect_flags(info->mti_exp) &
+			      OBD_CONNECT_LFSCK)) {
+				return -EIO;
+			} else {
+				ma->ma_lmm_size = rc;
+				ma->ma_valid |= MA_LOV;
+			}
 		} else if (strcmp(name, XATTR_NAME_LMV) == 0) {
+			if (info->mti_big_lmm_used)
+				ma->ma_lmv = info->mti_big_lmm;
+
 			ma->ma_lmv_size = rc;
 			ma->ma_valid |= MA_LMV;
 		} else if (strcmp(name, XATTR_NAME_DEFAULT_LMV) == 0) {
@@ -583,6 +746,10 @@ int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o,
 			ma->ma_valid |= MA_LMV_DEF;
 		}
 
+		/* Update mdt_max_mdsize so all clients will be aware that */
+		if (info->mti_mdt->mdt_max_mdsize < rc)
+			info->mti_mdt->mdt_max_mdsize = rc;
+
 		rc = 0;
 	} else if (rc == -ENODATA) {
 		/* no LOV EA */
@@ -595,23 +762,7 @@ int mdt_stripe_get(struct mdt_thread_info *info, struct mdt_object *o,
 		rc = mdt_big_xattr_get(info, o, name);
 		if (rc > 0) {
 			info->mti_big_lmm_used = 1;
-			if (!strcmp(name, XATTR_NAME_LOV)) {
-				ma->ma_valid |= MA_LOV;
-				ma->ma_lmm = info->mti_big_lmm;
-				ma->ma_lmm_size = rc;
-			} else if (!strcmp(name, XATTR_NAME_LMV)) {
-				ma->ma_valid |= MA_LMV;
-				ma->ma_lmv = info->mti_big_lmm;
-				ma->ma_lmv_size = rc;
-			} else {
-				return -EINVAL;
-			}
-
-			/* update mdt_max_mdsize so all clients
-			 * will be aware about that */
-			if (info->mti_mdt->mdt_max_mdsize < rc)
-				info->mti_mdt->mdt_max_mdsize = rc;
-			rc = 0;
+			goto got;
 		}
 	}
 
@@ -671,13 +822,17 @@ int mdt_attr_get_complex(struct mdt_thread_info *info,
 	const struct lu_env *env = info->mti_env;
 	struct md_object    *next = mdt_object_child(o);
 	struct lu_buf       *buf = &info->mti_buf;
-	u32                  mode = lu_object_attr(&next->mo_lu);
 	int                  need = ma->ma_need;
 	int                  rc = 0, rc2;
+	u32                  mode;
 	ENTRY;
 
 	ma->ma_valid = 0;
 
+	if (mdt_object_exists(o) == 0)
+		GOTO(out, rc = -ENOENT);
+	mode = lu_object_attr(&next->mo_lu);
+
 	if (need & MA_INODE) {
 		ma->ma_need = MA_INODE;
 		rc = mo_attr_get(env, next, ma);
@@ -712,19 +867,6 @@ int mdt_attr_get_complex(struct mdt_thread_info *info,
 			GOTO(out, rc);
 	}
 
-	if (need & MA_SOM && S_ISREG(mode)) {
-		buf->lb_buf = info->mti_xattr_buf;
-		buf->lb_len = sizeof(info->mti_xattr_buf);
-		CLASSERT(sizeof(struct som_attrs) <=
-			 sizeof(info->mti_xattr_buf));
-		rc2 = mo_xattr_get(info->mti_env, next, buf, XATTR_NAME_SOM);
-		rc2 = lustre_buf2som(info->mti_xattr_buf, rc2, ma->ma_som);
-		if (rc2 == 0)
-			ma->ma_valid |= MA_SOM;
-		else if (rc2 < 0 && rc2 != -ENODATA)
-			GOTO(out, rc = rc2);
-	}
-
 	if (need & MA_HSM && S_ISREG(mode)) {
 		buf->lb_buf = info->mti_xattr_buf;
 		buf->lb_len = sizeof(info->mti_xattr_buf);
@@ -763,38 +905,40 @@ out:
 static int mdt_getattr_internal(struct mdt_thread_info *info,
                                 struct mdt_object *o, int ma_need)
 {
-        struct md_object        *next = mdt_object_child(o);
-        const struct mdt_body   *reqbody = info->mti_body;
-        struct ptlrpc_request   *req = mdt_info_req(info);
-        struct md_attr          *ma = &info->mti_attr;
-        struct lu_attr          *la = &ma->ma_attr;
-        struct req_capsule      *pill = info->mti_pill;
-        const struct lu_env     *env = info->mti_env;
-        struct mdt_body         *repbody;
-        struct lu_buf           *buffer = &info->mti_buf;
-        int                     rc;
-	int			is_root;
-        ENTRY;
+	struct md_object	*next = mdt_object_child(o);
+	const struct mdt_body	*reqbody = info->mti_body;
+	struct ptlrpc_request	*req = mdt_info_req(info);
+	struct md_attr		*ma = &info->mti_attr;
+	struct lu_attr		*la = &ma->ma_attr;
+	struct req_capsule	*pill = info->mti_pill;
+	const struct lu_env	*env = info->mti_env;
+	struct mdt_body		*repbody;
+	struct lu_buf		*buffer = &info->mti_buf;
+	struct obd_export	*exp = info->mti_exp;
+	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
+	int			 rc;
+	int			 is_root;
+	ENTRY;
 
-        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK))
-                RETURN(err_serious(-ENOMEM));
+	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK))
+		RETURN(err_serious(-ENOMEM));
 
-        repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
+	repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
 
-        ma->ma_valid = 0;
+	ma->ma_valid = 0;
 
 	if (mdt_object_remote(o)) {
 		/* This object is located on remote node.*/
-		/* Return -EIO for old client */
+		/* Return -ENOTSUPP for old client */
 		if (!mdt_is_dne_client(req->rq_export))
-			GOTO(out, rc = -EIO);
+			GOTO(out, rc = -ENOTSUPP);
 
-		repbody->fid1 = *mdt_object_fid(o);
-		repbody->valid = OBD_MD_FLID | OBD_MD_MDS;
+		repbody->mbo_fid1 = *mdt_object_fid(o);
+		repbody->mbo_valid = OBD_MD_FLID | OBD_MD_MDS;
 		GOTO(out, rc = 0);
 	}
 
-	if (reqbody->eadatasize > 0) {
+	if (reqbody->mbo_eadatasize > 0) {
 		buffer->lb_buf = req_capsule_server_get(pill, &RMF_MDT_MD);
 		if (buffer->lb_buf == NULL)
 			GOTO(out, rc = -EPROTO);
@@ -811,15 +955,15 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 
 	/* If it is dir object and client require MEA, then we got MEA */
 	if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
-	    (reqbody->valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) {
+	    (reqbody->mbo_valid & (OBD_MD_MEA | OBD_MD_DEFAULT_MEA))) {
 		/* Assumption: MDT_MD size is enough for lmv size. */
 		ma->ma_lmv = buffer->lb_buf;
 		ma->ma_lmv_size = buffer->lb_len;
 		ma->ma_need = MA_INODE;
 		if (ma->ma_lmv_size > 0) {
-			if (reqbody->valid & OBD_MD_MEA)
+			if (reqbody->mbo_valid & OBD_MD_MEA)
 				ma->ma_need |= MA_LMV;
-			else if (reqbody->valid & OBD_MD_DEFAULT_MEA)
+			else if (reqbody->mbo_valid & OBD_MD_DEFAULT_MEA)
 				ma->ma_need |= MA_LMV_DEF;
 		}
 	} else {
@@ -831,14 +975,12 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 	}
 
         if (S_ISDIR(lu_object_attr(&next->mo_lu)) &&
-            reqbody->valid & OBD_MD_FLDIREA  &&
+	    reqbody->mbo_valid & OBD_MD_FLDIREA  &&
             lustre_msg_get_opc(req->rq_reqmsg) == MDS_GETATTR) {
                 /* get default stripe info for this dir. */
                 ma->ma_need |= MA_LOV_DEF;
         }
         ma->ma_need |= ma_need;
-        if (ma->ma_need & MA_SOM)
-                ma->ma_som = &info->mti_u.som.data;
 
 	rc = mdt_attr_get_complex(info, o, ma);
 	if (unlikely(rc)) {
@@ -849,10 +991,11 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 	}
 
 	/* if file is released, check if a restore is running */
-	if ((ma->ma_valid & MA_HSM) && (ma->ma_hsm.mh_flags & HS_RELEASED) &&
-	    mdt_hsm_restore_is_running(info, mdt_object_fid(o))) {
-		repbody->t_state = MS_RESTORE;
-		repbody->valid |= OBD_MD_TSTATE;
+	if (ma->ma_valid & MA_HSM) {
+		repbody->mbo_valid |= OBD_MD_TSTATE;
+		if ((ma->ma_hsm.mh_flags & HS_RELEASED) &&
+		    mdt_hsm_restore_is_running(info, mdt_object_fid(o)))
+			repbody->mbo_t_state = MS_RESTORE;
 	}
 
 	is_root = lu_fid_eq(mdt_object_fid(o), &info->mti_mdt->mdt_md_root_fid);
@@ -889,30 +1032,38 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
         if (mdt_body_has_lov(la, reqbody)) {
                 if (ma->ma_valid & MA_LOV) {
                         LASSERT(ma->ma_lmm_size);
-                        repbody->eadatasize = ma->ma_lmm_size;
-                        if (S_ISDIR(la->la_mode))
-                                repbody->valid |= OBD_MD_FLDIREA;
-                        else
-                                repbody->valid |= OBD_MD_FLEASIZE;
-                        mdt_dump_lmm(D_INFO, ma->ma_lmm, repbody->valid);
+			repbody->mbo_eadatasize = ma->ma_lmm_size;
+			if (S_ISDIR(la->la_mode))
+				repbody->mbo_valid |= OBD_MD_FLDIREA;
+			else
+				repbody->mbo_valid |= OBD_MD_FLEASIZE;
+			mdt_dump_lmm(D_INFO, ma->ma_lmm, repbody->mbo_valid);
                 }
 		if (ma->ma_valid & MA_LMV) {
+			/* Return -ENOTSUPP for old client */
+			if (!mdt_is_striped_client(req->rq_export))
+				RETURN(-ENOTSUPP);
+
 			LASSERT(S_ISDIR(la->la_mode));
 			mdt_dump_lmv(D_INFO, ma->ma_lmv);
-			repbody->eadatasize = ma->ma_lmv_size;
-			repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_MEA);
+			repbody->mbo_eadatasize = ma->ma_lmv_size;
+			repbody->mbo_valid |= (OBD_MD_FLDIREA|OBD_MD_MEA);
 		}
 		if (ma->ma_valid & MA_LMV_DEF) {
+			/* Return -ENOTSUPP for old client */
+			if (!mdt_is_striped_client(req->rq_export))
+				RETURN(-ENOTSUPP);
 			LASSERT(S_ISDIR(la->la_mode));
-			repbody->eadatasize = ma->ma_lmv_size;
-			repbody->valid |= (OBD_MD_FLDIREA|OBD_MD_DEFAULT_MEA);
+			repbody->mbo_eadatasize = ma->ma_lmv_size;
+			repbody->mbo_valid |= (OBD_MD_FLDIREA |
+					       OBD_MD_DEFAULT_MEA);
 		}
 	} else if (S_ISLNK(la->la_mode) &&
-		   reqbody->valid & OBD_MD_LINKNAME) {
+		   reqbody->mbo_valid & OBD_MD_LINKNAME) {
 		buffer->lb_buf = ma->ma_lmm;
 		/* eadatasize from client includes NULL-terminator, so
 		 * there is no need to read it */
-		buffer->lb_len = reqbody->eadatasize - 1;
+		buffer->lb_len = reqbody->mbo_eadatasize - 1;
 		rc = mo_readlink(env, next, buffer);
 		if (unlikely(rc <= 0)) {
 			CERROR("%s: readlink failed for "DFID": rc = %d\n",
@@ -924,16 +1075,16 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 
 			if (OBD_FAIL_CHECK(OBD_FAIL_MDS_READLINK_EPROTO))
 				rc -= 2;
-			repbody->valid |= OBD_MD_LINKNAME;
+			repbody->mbo_valid |= OBD_MD_LINKNAME;
 			/* we need to report back size with NULL-terminator
 			 * because client expects that */
-			repbody->eadatasize = rc + 1;
-			if (repbody->eadatasize != reqbody->eadatasize)
+			repbody->mbo_eadatasize = rc + 1;
+			if (repbody->mbo_eadatasize != reqbody->mbo_eadatasize)
 				CDEBUG(D_INODE, "%s: Read shorter symlink %d "
 				       "on "DFID ", expected %d\n",
 				       mdt_obd_name(info->mti_mdt),
 				       rc, PFID(mdt_object_fid(o)),
-				       reqbody->eadatasize - 1);
+				       reqbody->mbo_eadatasize - 1);
 			/* NULL terminate */
 			((char *)ma->ma_lmm)[rc] = 0;
 
@@ -947,75 +1098,34 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
                 }
         }
 
-        if (reqbody->valid & OBD_MD_FLMODEASIZE) {
-		repbody->max_cookiesize = 0;
-                repbody->max_mdsize = info->mti_mdt->mdt_max_mdsize;
-                repbody->valid |= OBD_MD_FLMODEASIZE;
-                CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & "
-                       "MAX_COOKIE to : %d:%d\n", repbody->max_mdsize,
-                       repbody->max_cookiesize);
-        }
+	if (reqbody->mbo_valid & OBD_MD_FLMODEASIZE) {
+		repbody->mbo_max_mdsize = info->mti_mdt->mdt_max_mdsize;
+		repbody->mbo_valid |= OBD_MD_FLMODEASIZE;
+		CDEBUG(D_INODE, "changing the max MD size to %u\n",
+		       repbody->mbo_max_mdsize);
+	}
 
-        if (exp_connect_rmtclient(info->mti_exp) &&
-            reqbody->valid & OBD_MD_FLRMTPERM) {
-                void *buf = req_capsule_server_get(pill, &RMF_ACL);
+	if (exp_connect_rmtclient(info->mti_exp) &&
+	    reqbody->mbo_valid & OBD_MD_FLRMTPERM) {
+		void *buf = req_capsule_server_get(pill, &RMF_ACL);
 
-                /* mdt_getattr_lock only */
-                rc = mdt_pack_remote_perm(info, o, buf);
-                if (rc) {
-                        repbody->valid &= ~OBD_MD_FLRMTPERM;
-                        repbody->aclsize = 0;
-                        RETURN(rc);
-                } else {
-                        repbody->valid |= OBD_MD_FLRMTPERM;
-                        repbody->aclsize = sizeof(struct mdt_remote_perm);
-                }
-        }
+		/* mdt_getattr_lock only */
+		rc = mdt_pack_remote_perm(info, o, buf);
+		if (rc) {
+			repbody->mbo_valid &= ~OBD_MD_FLRMTPERM;
+			repbody->mbo_aclsize = 0;
+			RETURN(rc);
+		} else {
+			repbody->mbo_valid |= OBD_MD_FLRMTPERM;
+			repbody->mbo_aclsize = sizeof(struct mdt_remote_perm);
+		}
+	}
 #ifdef CONFIG_FS_POSIX_ACL
 	else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) &&
-		 (reqbody->valid & OBD_MD_FLACL)) {
-                buffer->lb_buf = req_capsule_server_get(pill, &RMF_ACL);
-                buffer->lb_len = req_capsule_get_size(pill,
-                                                      &RMF_ACL, RCL_SERVER);
-                if (buffer->lb_len > 0) {
-                        rc = mo_xattr_get(env, next, buffer,
-                                          XATTR_NAME_ACL_ACCESS);
-                        if (rc < 0) {
-                                if (rc == -ENODATA) {
-                                        repbody->aclsize = 0;
-                                        repbody->valid |= OBD_MD_FLACL;
-                                        rc = 0;
-                                } else if (rc == -EOPNOTSUPP) {
-                                        rc = 0;
-				} else {
-					CERROR("%s: unable to read "DFID
-					       " ACL: rc = %d\n",
-					       mdt_obd_name(info->mti_mdt),
-					       PFID(mdt_object_fid(o)), rc);
-				}
-                        } else {
-                                repbody->aclsize = rc;
-                                repbody->valid |= OBD_MD_FLACL;
-                                rc = 0;
-                        }
-                }
-        }
+		 (reqbody->mbo_valid & OBD_MD_FLACL))
+		rc = mdt_pack_acl2body(info, repbody, o, nodemap);
 #endif
 
-	if (reqbody->valid & OBD_MD_FLMDSCAPA &&
-	    info->mti_mdt->mdt_lut.lut_mds_capa &&
-	    exp_connect_flags(info->mti_exp) & OBD_CONNECT_MDS_CAPA) {
-                struct lustre_capa *capa;
-
-                capa = req_capsule_server_get(pill, &RMF_CAPA1);
-                LASSERT(capa);
-                capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
-                rc = mo_capa_get(env, next, capa, 0);
-                if (rc)
-                        RETURN(rc);
-                repbody->valid |= OBD_MD_FLMDSCAPA;
-        }
-
 out:
         if (rc == 0)
 		mdt_counter_incr(req, LPROC_MDT_GETATTR);
@@ -1023,38 +1133,6 @@ out:
         RETURN(rc);
 }
 
-static int mdt_renew_capa(struct mdt_thread_info *info)
-{
-        struct mdt_object  *obj = info->mti_object;
-        struct mdt_body    *body;
-        struct lustre_capa *capa, *c;
-        int rc;
-        ENTRY;
-
-        /* if object doesn't exist, or server has disabled capability,
-         * return directly, client will find body->valid OBD_MD_FLOSSCAPA
-         * flag not set.
-         */
-	if (!obj || !info->mti_mdt->mdt_lut.lut_oss_capa ||
-	    !(exp_connect_flags(info->mti_exp) & OBD_CONNECT_OSS_CAPA))
-		RETURN(0);
-
-        body = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
-        LASSERT(body != NULL);
-
-        c = req_capsule_client_get(info->mti_pill, &RMF_CAPA1);
-        LASSERT(c);
-
-        capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
-        LASSERT(capa);
-
-        *capa = *c;
-        rc = mo_capa_get(info->mti_env, mdt_object_child(obj), capa, 1);
-        if (rc == 0)
-                body->valid |= OBD_MD_FLOSSCAPA;
-        RETURN(rc);
-}
-
 static int mdt_getattr(struct tgt_session_info *tsi)
 {
 	struct mdt_thread_info	*info = tsi2mdt_info(tsi);
@@ -1067,15 +1145,6 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 
         reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY);
         LASSERT(reqbody);
-
-        if (reqbody->valid & OBD_MD_FLOSSCAPA) {
-                rc = req_capsule_server_pack(pill);
-                if (unlikely(rc))
-                        RETURN(err_serious(rc));
-                rc = mdt_renew_capa(info);
-                GOTO(out_shrink, rc);
-        }
-
         LASSERT(obj != NULL);
 	LASSERT(lu_object_assert_exists(&obj->mot_obj));
 
@@ -1084,7 +1153,7 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 	 * guess at EA size by just reading it from disk.
 	 * Exceptions are readdir and (missing) directory striping */
 	/* Readlink */
-	if (reqbody->valid & OBD_MD_LINKNAME) {
+	if (reqbody->mbo_valid & OBD_MD_LINKNAME) {
 		/* No easy way to know how long is the symlink, but it cannot
 		 * be more than PATH_MAX, so we allocate +1 */
 		rc = PATH_MAX + 1;
@@ -1094,7 +1163,7 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 	 */
 	} else if (lu_fid_eq(mdt_object_fid(obj),
 			     &info->mti_mdt->mdt_md_root_fid) &&
-		   (reqbody->valid & OBD_MD_FLDIREA) &&
+		   (reqbody->mbo_valid & OBD_MD_FLDIREA) &&
 		   (lustre_msg_get_opc(mdt_info_req(info)->rq_reqmsg) ==
 								 MDS_GETATTR)) {
 		/* Should the default strping be bigger, mdt_fix_reply
@@ -1106,7 +1175,7 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 	}
 
 	if (rc < 0)
-		GOTO(out_shrink, rc);
+		GOTO(out, rc = err_serious(rc));
 
 	req_capsule_set_size(pill, &RMF_MDT_MD, RCL_SERVER, rc);
 
@@ -1116,25 +1185,20 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 
         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
         LASSERT(repbody != NULL);
-        repbody->eadatasize = 0;
-        repbody->aclsize = 0;
+	repbody->mbo_eadatasize = 0;
+	repbody->mbo_aclsize = 0;
 
-        if (reqbody->valid & OBD_MD_FLRMTPERM)
-                rc = mdt_init_ucred(info, reqbody);
-        else
-                rc = mdt_check_ucred(info);
-        if (unlikely(rc))
-                GOTO(out_shrink, rc);
+	if (reqbody->mbo_valid & OBD_MD_FLRMTPERM)
+		rc = mdt_init_ucred(info, reqbody);
+	else
+		rc = mdt_check_ucred(info);
+	if (unlikely(rc))
+		GOTO(out_shrink, rc);
 
-        info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
+	info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF);
 
-        /*
-         * Don't check capability at all, because rename might getattr for
-         * remote obj, and at that time no capability is available.
-         */
-        mdt_set_capainfo(info, 1, &reqbody->fid1, BYPASS_CAPA);
-        rc = mdt_getattr_internal(info, obj, 0);
-        if (reqbody->valid & OBD_MD_FLRMTPERM)
+	rc = mdt_getattr_internal(info, obj, 0);
+	if (reqbody->mbo_valid & OBD_MD_FLRMTPERM)
                 mdt_exit_ucred(info);
         EXIT;
 out_shrink:
@@ -1147,33 +1211,30 @@ out:
 	return rc;
 }
 
-static int mdt_is_subdir(struct tgt_session_info *tsi)
+/**
+ * Exchange MOF_LOV_CREATED flags between two objects after a
+ * layout swap. No assumption is made on whether o1 or o2 have
+ * created objects or not.
+ *
+ * \param[in,out] o1	First swap layout object
+ * \param[in,out] o2	Second swap layout object
+ */
+static void mdt_swap_lov_flag(struct mdt_object *o1, struct mdt_object *o2)
 {
-	struct mdt_thread_info	*info = tsi2mdt_info(tsi);
-        struct mdt_object     *o = info->mti_object;
-        struct req_capsule    *pill = info->mti_pill;
-        const struct mdt_body *body = info->mti_body;
-        struct mdt_body       *repbody;
-        int                    rc;
-        ENTRY;
+	__u64	o1_flags;
 
-        LASSERT(o != NULL);
+	mutex_lock(&o1->mot_lov_mutex);
+	mutex_lock(&o2->mot_lov_mutex);
 
-        repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
+	o1_flags = o1->mot_flags;
+	o1->mot_flags = (o1->mot_flags & ~MOF_LOV_CREATED) |
+			(o2->mot_flags & MOF_LOV_CREATED);
 
-	/*
-	 * We save last checked parent fid to @repbody->fid1 for remote
-	 * directory case.
-	 */
-	LASSERT(fid_is_sane(&body->fid2));
-	LASSERT(mdt_object_exists(o) && !mdt_object_remote(o));
-	rc = mdo_is_subdir(info->mti_env, mdt_object_child(o),
-			   &body->fid2, &repbody->fid1);
-	if (rc == 0 || rc == -EREMOTE)
-		repbody->valid |= OBD_MD_FLID;
+	o2->mot_flags = (o2->mot_flags & ~MOF_LOV_CREATED) |
+			(o1_flags & MOF_LOV_CREATED);
 
-	mdt_thread_info_fini(info);
-	RETURN(rc);
+	mutex_unlock(&o2->mot_lov_mutex);
+	mutex_unlock(&o1->mot_lov_mutex);
 }
 
 static int mdt_swap_layouts(struct tgt_session_info *tsi)
@@ -1200,28 +1261,18 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi)
 	info = tsi2mdt_info(tsi);
 
 	if (info->mti_dlm_req != NULL)
-		ldlm_request_cancel(req, info->mti_dlm_req, 0);
-
-	if (req_capsule_get_size(info->mti_pill, &RMF_CAPA1, RCL_CLIENT))
-		mdt_set_capainfo(info, 0, &info->mti_body->fid1,
-				 req_capsule_client_get(info->mti_pill,
-							&RMF_CAPA1));
-
-	if (req_capsule_get_size(info->mti_pill, &RMF_CAPA2, RCL_CLIENT))
-		mdt_set_capainfo(info, 1, &info->mti_body->fid2,
-				 req_capsule_client_get(info->mti_pill,
-							&RMF_CAPA2));
+		ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
 
 	o1 = info->mti_object;
 	o = o2 = mdt_object_find(info->mti_env, info->mti_mdt,
-				&info->mti_body->fid2);
+				&info->mti_body->mbo_fid2);
 	if (IS_ERR(o))
 		GOTO(out, rc = PTR_ERR(o));
 
 	if (mdt_object_remote(o) || !mdt_object_exists(o)) /* remote object */
 		GOTO(put, rc = -ENOENT);
 
-	rc = lu_fid_cmp(&info->mti_body->fid1, &info->mti_body->fid2);
+	rc = lu_fid_cmp(&info->mti_body->mbo_fid1, &info->mti_body->mbo_fid2);
 	if (unlikely(rc == 0)) /* same file, you kidding me? no-op. */
 		GOTO(put, rc);
 
@@ -1250,18 +1301,22 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi)
 	mdt_lock_reg_init(lh2, LCK_EX);
 
 	rc = mdt_object_lock(info, o1, lh1, MDS_INODELOCK_LAYOUT |
-			     MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK);
+			     MDS_INODELOCK_XATTR);
 	if (rc < 0)
 		GOTO(put, rc);
 
 	rc = mdt_object_lock(info, o2, lh2, MDS_INODELOCK_LAYOUT |
-			     MDS_INODELOCK_XATTR, MDT_LOCAL_LOCK);
+			     MDS_INODELOCK_XATTR);
 	if (rc < 0)
 		GOTO(unlock1, rc);
 
 	rc = mo_swap_layouts(info->mti_env, mdt_object_child(o1),
 			     mdt_object_child(o2), msl->msl_flags);
-	GOTO(unlock2, rc);
+	if (rc < 0)
+		GOTO(unlock2, rc);
+
+	mdt_swap_lov_flag(o1, o2);
+
 unlock2:
 	mdt_object_unlock(info, o2, lh2, rc);
 unlock1:
@@ -1274,47 +1329,36 @@ out:
 }
 
 static int mdt_raw_lookup(struct mdt_thread_info *info,
-                          struct mdt_object *parent,
-                          const struct lu_name *lname,
-                          struct ldlm_reply *ldlm_rep)
-{
-        struct md_object *next = mdt_object_child(info->mti_object);
-        const struct mdt_body *reqbody = info->mti_body;
-        struct lu_fid *child_fid = &info->mti_tmp_fid1;
-        struct mdt_body *repbody;
-        int rc;
-        ENTRY;
-
-        if (reqbody->valid != OBD_MD_FLID)
-                RETURN(0);
-
-        LASSERT(!info->mti_cross_ref);
+			  struct mdt_object *parent,
+			  const struct lu_name *lname,
+			  struct ldlm_reply *ldlm_rep)
+{
+	struct lu_fid	*child_fid = &info->mti_tmp_fid1;
+	int		 rc;
+	ENTRY;
 
-        /* Only got the fid of this obj by name */
-        fid_zero(child_fid);
-        rc = mdo_lookup(info->mti_env, next, lname, child_fid,
-                        &info->mti_spec);
-#if 0
-        /* XXX is raw_lookup possible as intent operation? */
-        if (rc != 0) {
-                if (rc == -ENOENT)
-                        mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
-                RETURN(rc);
-        } else
-                mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
+	LASSERT(!info->mti_cross_ref);
+
+	/* Only got the fid of this obj by name */
+	fid_zero(child_fid);
+	rc = mdo_lookup(info->mti_env, mdt_object_child(info->mti_object),
+			lname, child_fid, &info->mti_spec);
+	if (rc == 0) {
+		struct mdt_body *repbody;
+
+		repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
+		repbody->mbo_fid1 = *child_fid;
+		repbody->mbo_valid = OBD_MD_FLID;
+		mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
+	} else if (rc == -ENOENT) {
+		mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
+	}
 
-        repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
-#endif
-        if (rc == 0) {
-                repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
-                repbody->fid1 = *child_fid;
-                repbody->valid = OBD_MD_FLID;
-        }
-        RETURN(1);
+	RETURN(rc);
 }
 
 /*
- * UPDATE lock should be taken against parent, and be release before exit;
+ * UPDATE lock should be taken against parent, and be released before exit;
  * child_bits lock should be taken against child, and be returned back:
  *            (1)normal request should release the child lock;
  *            (2)intent request will grant the lock to client.
@@ -1328,7 +1372,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
         struct mdt_body        *reqbody   = NULL;
         struct mdt_object      *parent    = info->mti_object;
         struct mdt_object      *child;
-        struct md_object       *next      = mdt_object_child(parent);
         struct lu_fid          *child_fid = &info->mti_tmp_fid1;
         struct lu_name         *lname     = NULL;
         struct mdt_lock_handle *lhp       = NULL;
@@ -1343,7 +1386,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 	LASSERT(ergo(is_resent,
 		     lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT));
 
-	LASSERT(parent != NULL);
+	if (parent == NULL)
+		RETURN(-ENOENT);
 
 	if (info->mti_cross_ref) {
 		/* Only getattr on the child. Parent is on another node. */
@@ -1354,15 +1398,10 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 		       "ldlm_rep = %p\n",
 		       PFID(mdt_object_fid(child)), ldlm_rep);
 
-		if (is_resent) {
-			/* Do not take lock for resent case. */
-			lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
-			LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
-				 lhc->mlh_reg_lh.cookie);
-			LASSERT(fid_res_name_eq(mdt_object_fid(child),
-						&lock->l_resource->lr_name));
-			LDLM_LOCK_PUT(lock);
-		} else {
+		rc = mdt_check_resent_lock(info, child, lhc);
+		if (rc < 0) {
+			RETURN(rc);
+		} else if (rc > 0) {
 			mdt_lock_handle_init(lhc);
 			mdt_lock_reg_init(lhc, LCK_PR);
 
@@ -1374,8 +1413,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 					MDS_INODELOCK_LAYOUT);
 			child_bits |= MDS_INODELOCK_PERM | MDS_INODELOCK_UPDATE;
 
-			rc = mdt_object_lock(info, child, lhc, child_bits,
-					     MDT_LOCAL_LOCK);
+			rc = mdt_object_lock(info, child, lhc, child_bits);
 			if (rc < 0)
 				RETURN(rc);
 		}
@@ -1389,7 +1427,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 			RETURN(-ENOENT);
 		}
 
-		mdt_set_capainfo(info, 0, mdt_object_fid(child), BYPASS_CAPA);
 		rc = mdt_getattr_internal(info, child, 0);
 		if (unlikely(rc != 0))
 			mdt_object_unlock(info, child, lhc, 1);
@@ -1409,7 +1446,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 		if (unlikely(reqbody == NULL))
 			RETURN(err_serious(-EPROTO));
 
-		*child_fid = reqbody->fid2;
+		*child_fid = reqbody->mbo_fid2;
 
 		if (unlikely(!fid_is_sane(child_fid)))
 			RETURN(err_serious(-EINVAL));
@@ -1417,7 +1454,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 		CDEBUG(D_INODE, "getattr with lock for "DFID"/"DFID", "
 		       "ldlm_rep = %p\n",
 		       PFID(mdt_object_fid(parent)),
-		       PFID(&reqbody->fid2), ldlm_rep);
+		       PFID(&reqbody->mbo_fid2), ldlm_rep);
 	}
 
 	mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_EXECD);
@@ -1437,10 +1474,15 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 	}
 
 	if (lu_name_is_valid(lname)) {
-		rc = mdt_raw_lookup(info, parent, lname, ldlm_rep);
-		if (rc != 0) {
-			if (rc > 0)
-				rc = 0;
+		/* Always allow to lookup ".." */
+		if (unlikely(lname->ln_namelen == 2 &&
+			     lname->ln_name[0] == '.' &&
+			     lname->ln_name[1] == '.'))
+			info->mti_spec.sp_permitted = 1;
+
+		if (info->mti_body->mbo_valid == OBD_MD_FLID) {
+			rc = mdt_raw_lookup(info, parent, lname, ldlm_rep);
+
 			RETURN(rc);
 		}
 
@@ -1448,17 +1490,16 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 		if (S_ISDIR(lu_object_attr(&parent->mot_obj))) {
 			lhp = &info->mti_lh[MDT_LH_PARENT];
 			mdt_lock_pdo_init(lhp, LCK_PR, lname);
-                        rc = mdt_object_lock(info, parent, lhp,
-                                             MDS_INODELOCK_UPDATE,
-                                             MDT_LOCAL_LOCK);
-                        if (unlikely(rc != 0))
-                                RETURN(rc);
-                }
+			rc = mdt_object_lock(info, parent, lhp,
+					     MDS_INODELOCK_UPDATE);
+			if (unlikely(rc != 0))
+				RETURN(rc);
+		}
 
                 /* step 2: lookup child's fid by name */
                 fid_zero(child_fid);
-                rc = mdo_lookup(info->mti_env, next, lname, child_fid,
-                                &info->mti_spec);
+		rc = mdo_lookup(info->mti_env, mdt_object_child(parent), lname,
+				child_fid, &info->mti_spec);
 		if (rc == -ENOENT)
 			mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
 
@@ -1487,42 +1528,23 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 
 	if (unlikely(IS_ERR(child)))
 		GOTO(out_parent, rc = PTR_ERR(child));
-	if (is_resent) {
-		/* Do not take lock for resent case. */
-		lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
-		LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
-			 lhc->mlh_reg_lh.cookie);
 
-		if (!fid_res_name_eq(mdt_object_fid(child),
-				     &lock->l_resource->lr_name)) {
-			LASSERTF(fid_res_name_eq(mdt_object_fid(parent),
-						 &lock->l_resource->lr_name),
-				 "Lock res_id: "DLDLMRES", fid: "DFID"\n",
-				 PLDLMRES(lock->l_resource),
-				 PFID(mdt_object_fid(parent)));
-			CWARN("Although resent, but still not get child lock"
-			      "parent:"DFID" child:"DFID"\n",
-			      PFID(mdt_object_fid(parent)),
-			      PFID(mdt_object_fid(child)));
-			lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
-			LDLM_LOCK_PUT(lock);
-			GOTO(relock, 0);
-		}
-		LDLM_LOCK_PUT(lock);
-	} else {
-relock:
-                OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2);
+	OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout * 2);
+	if (!mdt_object_exists(child)) {
+		LU_OBJECT_DEBUG(D_INODE, info->mti_env,
+				&child->mot_obj,
+				"Object doesn't exist!\n");
+		GOTO(out_child, rc = -ENOENT);
+	}
+
+	rc = mdt_check_resent_lock(info, child, lhc);
+	if (rc < 0) {
+		GOTO(out_child, rc);
+	} else if (rc > 0) {
                 mdt_lock_handle_init(lhc);
 		mdt_lock_reg_init(lhc, LCK_PR);
 		try_layout = false;
 
-		if (!mdt_object_exists(child)) {
-			LU_OBJECT_DEBUG(D_INODE, info->mti_env,
-					&child->mot_obj,
-					"Object doesn't exist!\n");
-			GOTO(out_child, rc = -ENOENT);
-		}
-
 		if (!(child_bits & MDS_INODELOCK_UPDATE) &&
 		      mdt_object_exists(child) && !mdt_object_remote(child)) {
                         struct md_attr *ma = &info->mti_attr;
@@ -1561,12 +1583,12 @@ relock:
 			child_bits |= MDS_INODELOCK_LAYOUT;
 			/* try layout lock, it may fail to be granted due to
 			 * contention at LOOKUP or UPDATE */
-			if (!mdt_object_lock_try(info, child, lhc, child_bits,
-						 MDT_CROSS_LOCK)) {
+			if (!mdt_object_lock_try(info, child, lhc,
+						 child_bits)) {
 				child_bits &= ~MDS_INODELOCK_LAYOUT;
 				LASSERT(child_bits != 0);
 				rc = mdt_object_lock(info, child, lhc,
-						child_bits, MDT_CROSS_LOCK);
+						     child_bits);
 			} else {
 				ma_need |= MA_LOV;
 			}
@@ -1575,22 +1597,15 @@ relock:
 			 * client will enqueue the lock to the remote MDT */
 			if (mdt_object_remote(child))
 				child_bits &= ~MDS_INODELOCK_UPDATE;
-			rc = mdt_object_lock(info, child, lhc, child_bits,
-						MDT_CROSS_LOCK);
+			rc = mdt_object_lock(info, child, lhc, child_bits);
 		}
                 if (unlikely(rc != 0))
                         GOTO(out_child, rc);
         }
 
         lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
-        /* Get MA_SOM attributes if update lock is given. */
-        if (lock &&
-            lock->l_policy_data.l_inodebits.bits & MDS_INODELOCK_UPDATE &&
-            S_ISREG(lu_object_attr(&mdt_object_child(child)->mo_lu)))
-                ma_need |= MA_SOM;
 
         /* finally, we can get attr for child. */
-        mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
         rc = mdt_getattr_internal(info, child, ma_need);
         if (unlikely(rc != 0)) {
                 mdt_object_unlock(info, child, lhc, 1);
@@ -1602,8 +1617,6 @@ relock:
 			 "Lock res_id: "DLDLMRES", fid: "DFID"\n",
 			 PLDLMRES(lock->l_resource),
 			 PFID(mdt_object_fid(child)));
-		if (mdt_object_exists(child) && !mdt_object_remote(child))
-			mdt_pack_size2body(info, child);
         }
         if (lock)
                 LDLM_LOCK_PUT(lock);
@@ -1632,11 +1645,11 @@ static int mdt_getattr_name(struct tgt_session_info *tsi)
         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
         LASSERT(repbody != NULL);
 
-        info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
-        repbody->eadatasize = 0;
-        repbody->aclsize = 0;
+	info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF);
+	repbody->mbo_eadatasize = 0;
+	repbody->mbo_aclsize = 0;
 
-        rc = mdt_init_ucred(info, reqbody);
+        rc = mdt_init_ucred_intent_getattr(info, reqbody);
         if (unlikely(rc))
                 GOTO(out_shrink, rc);
 
@@ -1657,7 +1670,7 @@ out_shrink:
 }
 
 static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
-                         void *karg, void *uarg);
+			 void *karg, void __user *uarg);
 
 static int mdt_set_info(struct tgt_session_info *tsi)
 {
@@ -1711,6 +1724,9 @@ static int mdt_set_info(struct tgt_session_info *tsi)
 
 		rc = mdt_iocontrol(OBD_IOC_CHANGELOG_CLEAR, req->rq_export,
 				   vallen, val, NULL);
+	} else if (KEY_IS(KEY_EVICT_BY_NID)) {
+		if (vallen > 0)
+			obd_export_evict_by_nid(req->rq_export->exp_obd, val);
 	} else {
 		RETURN(-EINVAL);
 	}
@@ -1741,17 +1757,17 @@ static int mdt_readpage(struct tgt_session_info *tsi)
          * reqbody->size contains offset of where to start to read and
          * reqbody->nlink contains number bytes to read.
          */
-        rdpg->rp_hash = reqbody->size;
-        if (rdpg->rp_hash != reqbody->size) {
-                CERROR("Invalid hash: "LPX64" != "LPX64"\n",
-                       rdpg->rp_hash, reqbody->size);
-                RETURN(-EFAULT);
-        }
+	rdpg->rp_hash = reqbody->mbo_size;
+	if (rdpg->rp_hash != reqbody->mbo_size) {
+		CERROR("Invalid hash: "LPX64" != "LPX64"\n",
+		       rdpg->rp_hash, reqbody->mbo_size);
+		RETURN(-EFAULT);
+	}
 
-        rdpg->rp_attrs = reqbody->mode;
+	rdpg->rp_attrs = reqbody->mbo_mode;
 	if (exp_connect_flags(tsi->tsi_exp) & OBD_CONNECT_64BITHASH)
 		rdpg->rp_attrs |= LUDA_64BITHASH;
-	rdpg->rp_count  = min_t(unsigned int, reqbody->nlink,
+	rdpg->rp_count  = min_t(unsigned int, reqbody->mbo_nlink,
 				exp_max_brw_size(tsi->tsi_exp));
 	rdpg->rp_npages = (rdpg->rp_count + PAGE_CACHE_SIZE - 1) >>
 			  PAGE_CACHE_SHIFT;
@@ -1821,8 +1837,8 @@ static int mdt_reint_internal(struct mdt_thread_info *info,
         if (req_capsule_has_field(pill, &RMF_MDT_BODY, RCL_SERVER)) {
                 repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
                 LASSERT(repbody);
-                repbody->eadatasize = 0;
-                repbody->aclsize = 0;
+		repbody->mbo_eadatasize = 0;
+		repbody->mbo_aclsize = 0;
         }
 
         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_REINT_DELAY, 10);
@@ -1840,7 +1856,10 @@ static int mdt_reint_internal(struct mdt_thread_info *info,
         if (rc != 0)
                 GOTO(out_ucred, rc = err_serious(rc));
 
-        if (mdt_check_resent(info, mdt_reconstruct, lhc)) {
+	rc = mdt_check_resent(info, mdt_reconstruct, lhc);
+	if (rc < 0) {
+		GOTO(out_ucred, rc);
+	} else if (rc == 1) {
 		DEBUG_REQ(D_INODE, mdt_info_req(info), "resent opt.");
 		rc = lustre_msg_get_status(mdt_info_req(info)->rq_repmsg);
                 GOTO(out_ucred, rc);
@@ -1932,19 +1951,20 @@ static int mdt_device_sync(const struct lu_env *env, struct mdt_device *mdt)
 /* this should sync this object */
 static int mdt_object_sync(struct mdt_thread_info *info)
 {
-        struct md_object *next;
-        int rc;
-        ENTRY;
+	struct md_object *next;
+	int rc;
+	ENTRY;
 
-        if (!mdt_object_exists(info->mti_object)) {
-                CWARN("Non existing object  "DFID"!\n",
-                      PFID(mdt_object_fid(info->mti_object)));
-                RETURN(-ESTALE);
-        }
-        next = mdt_object_child(info->mti_object);
-        rc = mo_object_sync(info->mti_env, next);
+	if (!mdt_object_exists(info->mti_object)) {
+		CWARN("%s: non existing object "DFID": rc = %d\n",
+		      mdt_obd_name(info->mti_mdt),
+		      PFID(mdt_object_fid(info->mti_object)), -ESTALE);
+		RETURN(-ESTALE);
+	}
+	next = mdt_object_child(info->mti_object);
+	rc = mo_object_sync(info->mti_env, next);
 
-        RETURN(rc);
+	RETURN(rc);
 }
 
 static int mdt_sync(struct tgt_session_info *tsi)
@@ -1959,7 +1979,7 @@ static int mdt_sync(struct tgt_session_info *tsi)
 	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_SYNC_PACK))
 		RETURN(err_serious(-ENOMEM));
 
-	if (fid_seq(&tsi->tsi_mdt_body->fid1) == 0) {
+	if (fid_seq(&tsi->tsi_mdt_body->mbo_fid1) == 0) {
 		rc = mdt_device_sync(tsi->tsi_env, mdt_exp2dev(tsi->tsi_exp));
 	} else {
 		struct mdt_thread_info *info = tsi2mdt_info(tsi);
@@ -2001,6 +2021,7 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 	int			 id, rc;
 	struct mdt_device	*mdt = mdt_exp2dev(exp);
 	struct lu_device	*qmt = mdt->mdt_qmt_dev;
+	struct lu_nodemap	*nodemap = exp->exp_target_data.ted_nodemap;
 	ENTRY;
 
 	oqctl = req_capsule_client_get(pill, &RMF_OBD_QUOTACTL);
@@ -2012,18 +2033,12 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 		RETURN(err_serious(rc));
 
 	switch (oqctl->qc_cmd) {
-	case Q_QUOTACHECK:
-	case LUSTRE_Q_INVALIDATE:
-	case LUSTRE_Q_FINVALIDATE:
-	case Q_QUOTAON:
-	case Q_QUOTAOFF:
-	case Q_INITQUOTA:
-		/* deprecated, not used any more */
-		RETURN(-EOPNOTSUPP);
 		/* master quotactl */
-	case Q_GETINFO:
 	case Q_SETINFO:
 	case Q_SETQUOTA:
+		if (!nodemap_can_setquota(nodemap))
+			RETURN(-EPERM);
+	case Q_GETINFO:
 	case Q_GETQUOTA:
 		if (qmt == NULL)
 			RETURN(-EOPNOTSUPP);
@@ -2062,6 +2077,13 @@ static int mdt_quotactl(struct tgt_session_info *tsi)
 		}
 	}
 
+	if (oqctl->qc_type == USRQUOTA)
+		id = nodemap_map_id(nodemap, NODEMAP_UID,
+				    NODEMAP_CLIENT_TO_FS, id);
+	else if (oqctl->qc_type == GRPQUOTA)
+		id = nodemap_map_id(nodemap, NODEMAP_UID,
+				    NODEMAP_CLIENT_TO_FS, id);
+
 	repoqc = req_capsule_server_get(pill, &RMF_OBD_QUOTACTL);
 	if (repoqc == NULL)
 		RETURN(err_serious(-EFAULT));
@@ -2228,12 +2250,16 @@ struct mdt_object *mdt_object_find(const struct lu_env *env,
 static void mdt_device_commit_async(const struct lu_env *env,
                                     struct mdt_device *mdt)
 {
-        struct dt_device *dt = mdt->mdt_bottom;
-        int rc;
+	struct dt_device *dt = mdt->mdt_bottom;
+	int rc;
+	ENTRY;
 
-        rc = dt->dd_ops->dt_commit_async(env, dt);
-        if (unlikely(rc != 0))
-                CWARN("async commit start failed with rc = %d", rc);
+	rc = dt->dd_ops->dt_commit_async(env, dt);
+	if (unlikely(rc != 0))
+		CWARN("%s: async commit start failed: rc = %d\n",
+		      mdt_obd_name(mdt), rc);
+	atomic_inc(&mdt->mdt_async_commit_count);
+	EXIT;
 }
 
 /**
@@ -2289,17 +2315,22 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 
         if (flag == LDLM_CB_CANCELING)
                 RETURN(0);
+
         lock_res_and_lock(lock);
         if (lock->l_blocking_ast != mdt_blocking_ast) {
                 unlock_res_and_lock(lock);
                 RETURN(0);
         }
-        if (mdt_cos_is_enabled(mdt) &&
-            lock->l_req_mode & (LCK_PW | LCK_EX) &&
-            lock->l_blocking_lock != NULL &&
-            lock->l_client_cookie != lock->l_blocking_lock->l_client_cookie) {
-                mdt_set_lock_sync(lock);
-        }
+	if (lock->l_req_mode & (LCK_PW | LCK_EX) &&
+	    lock->l_blocking_lock != NULL) {
+		if (mdt_cos_is_enabled(mdt) &&
+		    lock->l_client_cookie !=
+		    lock->l_blocking_lock->l_client_cookie)
+			mdt_set_lock_sync(lock);
+		else if (mdt_slc_is_enabled(mdt) &&
+			 ldlm_is_cos_incompat(lock->l_blocking_lock))
+			mdt_set_lock_sync(lock);
+	}
         rc = ldlm_blocking_ast_nocheck(lock);
 
         /* There is no lock conflict if l_blocking_lock == NULL,
@@ -2308,10 +2339,11 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
         if (lock->l_req_mode == LCK_COS && lock->l_blocking_lock != NULL) {
                 struct lu_env env;
 
-                rc = lu_env_init(&env, LCT_LOCAL);
-                if (unlikely(rc != 0))
-                        CWARN("lu_env initialization failed with rc = %d,"
-                              "cannot start asynchronous commit\n", rc);
+		rc = lu_env_init(&env, LCT_LOCAL);
+		if (unlikely(rc != 0))
+			CWARN("%s: lu_env initialization failed, cannot "
+			      "start asynchronous commit: rc = %d\n",
+			      obd->obd_name, rc);
                 else
                         mdt_device_commit_async(&env, mdt);
                 lu_env_fini(&env);
@@ -2319,17 +2351,30 @@ int mdt_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
         RETURN(rc);
 }
 
-/* Used for cross-MDT lock */
+/*
+ * Blocking AST for cross-MDT lock
+ *
+ * Discard lock from uncommitted_slc_locks and cancel it.
+ *
+ * \param lock	the lock which blocks a request or cancelling lock
+ * \param desc	unused
+ * \param data	unused
+ * \param flag	indicates whether this cancelling or blocking callback
+ * \retval	0 on success
+ * \retval	negative number on error
+ */
 int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 			    void *data, int flag)
 {
 	struct lustre_handle lockh;
 	int		  rc;
+	ENTRY;
 
 	switch (flag) {
 	case LDLM_CB_BLOCKING:
 		ldlm_lock2handle(lock, &lockh);
-		rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
+		rc = ldlm_cli_cancel(&lockh,
+			ldlm_is_atomic_cb(lock) ? 0 : LCF_ASYNC);
 		if (rc < 0) {
 			CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
 			RETURN(rc);
@@ -2337,28 +2382,64 @@ int mdt_remote_blocking_ast(struct ldlm_lock *lock, struct ldlm_lock_desc *desc,
 		break;
 	case LDLM_CB_CANCELING:
 		LDLM_DEBUG(lock, "Revoke remote lock\n");
+		/* discard slc lock here so that it can be cleaned anytime,
+		 * especially for cleanup_resource() */
+		tgt_discard_slc_lock(lock);
 		break;
 	default:
 		LBUG();
 	}
+
 	RETURN(0);
 }
 
-int mdt_remote_object_lock(struct mdt_thread_info *mti,
-			   struct mdt_object *o, const struct lu_fid *fid,
-			   struct lustre_handle *lh, ldlm_mode_t mode,
-			   __u64 ibits)
+int mdt_check_resent_lock(struct mdt_thread_info *info,
+			  struct mdt_object *mo,
+			  struct mdt_lock_handle *lhc)
+{
+	/* the lock might already be gotten in ldlm_handle_enqueue() */
+	if (unlikely(lustre_handle_is_used(&lhc->mlh_reg_lh))) {
+		struct ptlrpc_request *req = mdt_info_req(info);
+		struct ldlm_lock      *lock;
+
+		lock = ldlm_handle2lock(&lhc->mlh_reg_lh);
+		LASSERT(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT);
+		if (lock == NULL) {
+			/* Lock is pinned by ldlm_handle_enqueue0() as it is
+			 * a resend case, however, it could be already destroyed
+			 * due to client eviction or a raced cancel RPC. */
+			LDLM_DEBUG_NOLOCK("Invalid lock handle "LPX64"\n",
+					  lhc->mlh_reg_lh.cookie);
+			RETURN(-ESTALE);
+		}
+
+		if (!fid_res_name_eq(mdt_object_fid(mo),
+				     &lock->l_resource->lr_name)) {
+			CWARN("%s: Although resent, but still not "
+			      "get child lock:"DFID"\n",
+			      info->mti_exp->exp_obd->obd_name,
+			      PFID(mdt_object_fid(mo)));
+			LDLM_LOCK_PUT(lock);
+			RETURN(-EPROTO);
+		}
+		LDLM_LOCK_PUT(lock);
+		return 0;
+	}
+	return 1;
+}
+
+int mdt_remote_object_lock(struct mdt_thread_info *mti, struct mdt_object *o,
+			   const struct lu_fid *fid, struct lustre_handle *lh,
+			   enum ldlm_mode mode, __u64 ibits, bool nonblock)
 {
 	struct ldlm_enqueue_info *einfo = &mti->mti_einfo;
-	ldlm_policy_data_t *policy = &mti->mti_policy;
+	union ldlm_policy_data *policy = &mti->mti_policy;
 	struct ldlm_res_id *res_id = &mti->mti_res_id;
 	int rc = 0;
 	ENTRY;
 
 	LASSERT(mdt_object_remote(o));
 
-	LASSERT(ibits == MDS_INODELOCK_UPDATE);
-
 	fid_build_reg_res_name(fid, res_id);
 
 	memset(einfo, 0, sizeof(*einfo));
@@ -2368,6 +2449,8 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti,
 	einfo->ei_cb_cp = ldlm_completion_ast;
 	einfo->ei_enq_slave = 0;
 	einfo->ei_res_id = res_id;
+	if (nonblock)
+		einfo->ei_nonblock = 1;
 
 	memset(policy, 0, sizeof(*policy));
 	policy->l_inodebits.bits = ibits;
@@ -2380,20 +2463,28 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti,
 static int mdt_object_local_lock(struct mdt_thread_info *info,
 				 struct mdt_object *o,
 				 struct mdt_lock_handle *lh, __u64 ibits,
-				 bool nonblock, int locality)
+				 bool nonblock, bool cos_incompat)
 {
-        struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
-        ldlm_policy_data_t *policy = &info->mti_policy;
-        struct ldlm_res_id *res_id = &info->mti_res_id;
-	__u64 dlmflags;
-        int rc;
-        ENTRY;
+	struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
+	union ldlm_policy_data *policy = &info->mti_policy;
+	struct ldlm_res_id *res_id = &info->mti_res_id;
+	__u64 dlmflags = 0;
+	int rc;
+	ENTRY;
 
         LASSERT(!lustre_handle_is_used(&lh->mlh_reg_lh));
         LASSERT(!lustre_handle_is_used(&lh->mlh_pdo_lh));
         LASSERT(lh->mlh_reg_mode != LCK_MINMODE);
         LASSERT(lh->mlh_type != MDT_NUL_LOCK);
 
+	if (cos_incompat) {
+		LASSERT(lh->mlh_reg_mode == LCK_PW ||
+			lh->mlh_reg_mode == LCK_EX);
+		dlmflags |= LDLM_FL_COS_INCOMPAT;
+	} else if (mdt_cos_is_enabled(info->mti_mdt)) {
+		dlmflags |= LDLM_FL_COS_ENABLED;
+	}
+
 	/* Only enqueue LOOKUP lock for remote object */
 	if (mdt_object_remote(o))
 		LASSERT(ibits == MDS_INODELOCK_LOOKUP);
@@ -2413,7 +2504,7 @@ static int mdt_object_local_lock(struct mdt_thread_info *info,
         memset(policy, 0, sizeof(*policy));
         fid_build_reg_res_name(mdt_object_fid(o), res_id);
 
-	dlmflags = LDLM_FL_ATOMIC_CB;
+	dlmflags |= LDLM_FL_ATOMIC_CB;
 	if (nonblock)
 		dlmflags |= LDLM_FL_BLOCK_NOWAIT;
 
@@ -2435,8 +2526,8 @@ static int mdt_object_local_lock(struct mdt_thread_info *info,
 					  policy, res_id, dlmflags,
 					  info->mti_exp == NULL ? NULL :
 					  &info->mti_exp->exp_handle.h_cookie);
-                        if (unlikely(rc))
-                                RETURN(rc);
+			if (unlikely(rc != 0))
+				GOTO(out_unlock, rc);
                 }
 
                 /*
@@ -2457,39 +2548,46 @@ static int mdt_object_local_lock(struct mdt_thread_info *info,
 			  res_id, LDLM_FL_LOCAL_ONLY | dlmflags,
 			  info->mti_exp == NULL ? NULL :
 			  &info->mti_exp->exp_handle.h_cookie);
-        if (rc)
-                mdt_object_unlock(info, o, lh, 1);
-        else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) &&
-                 lh->mlh_pdo_hash != 0 &&
-                 (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX)) {
-                OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15);
-        }
+out_unlock:
+	if (rc != 0)
+		mdt_object_unlock(info, o, lh, 1);
+	else if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_PDO_LOCK)) &&
+		   lh->mlh_pdo_hash != 0 &&
+		   (lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX))
+		OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK, 15);
 
-        RETURN(rc);
+	RETURN(rc);
 }
 
 static int
 mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o,
-			 struct mdt_lock_handle *lh, __u64 ibits,
-			 bool nonblock, int locality)
+			 struct mdt_lock_handle *lh, __u64 ibits, bool nonblock,
+			 bool cos_incompat)
 {
+	struct mdt_lock_handle *local_lh = NULL;
 	int rc;
 	ENTRY;
 
-	if (!mdt_object_remote(o))
-		return mdt_object_local_lock(info, o, lh, ibits, nonblock,
-					     locality);
-
-	if (locality == MDT_LOCAL_LOCK) {
-		CERROR("%s: try to get local lock for remote object"
-		       DFID".\n", mdt_obd_name(info->mti_mdt),
-		       PFID(mdt_object_fid(o)));
-		RETURN(-EPROTO);
+	if (!mdt_object_remote(o)) {
+		rc = mdt_object_local_lock(info, o, lh, ibits, nonblock,
+					   cos_incompat);
+		RETURN(rc);
 	}
 
 	/* XXX do not support PERM/LAYOUT/XATTR lock for remote object yet */
 	ibits &= ~(MDS_INODELOCK_PERM | MDS_INODELOCK_LAYOUT |
 		   MDS_INODELOCK_XATTR);
+
+	/* Only enqueue LOOKUP lock for remote object */
+	if (ibits & MDS_INODELOCK_LOOKUP) {
+		rc = mdt_object_local_lock(info, o, lh, MDS_INODELOCK_LOOKUP,
+					   nonblock, cos_incompat);
+		if (rc != ELDLM_OK)
+			RETURN(rc);
+
+		local_lh = lh;
+	}
+
 	if (ibits & MDS_INODELOCK_UPDATE) {
 		/* Sigh, PDO needs to enqueue 2 locks right now, but
 		 * enqueue RPC can only request 1 lock, to avoid extra
@@ -2507,36 +2605,54 @@ mdt_object_lock_internal(struct mdt_thread_info *info, struct mdt_object *o,
 		rc = mdt_remote_object_lock(info, o, mdt_object_fid(o),
 					    &lh->mlh_rreg_lh,
 					    lh->mlh_rreg_mode,
-					    MDS_INODELOCK_UPDATE);
-		if (rc != ELDLM_OK)
-			RETURN(rc);
-	}
-
-	/* Only enqueue LOOKUP lock for remote object */
-	if (ibits & MDS_INODELOCK_LOOKUP) {
-		rc = mdt_object_local_lock(info, o, lh,
-					   MDS_INODELOCK_LOOKUP,
-					   nonblock, locality);
-		if (rc != ELDLM_OK)
+					    MDS_INODELOCK_UPDATE, nonblock);
+		if (rc != ELDLM_OK) {
+			if (local_lh != NULL)
+				mdt_object_unlock(info, o, local_lh, rc);
 			RETURN(rc);
+		}
 	}
 
 	RETURN(0);
 }
 
 int mdt_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
-		    struct mdt_lock_handle *lh, __u64 ibits, int locality)
+		    struct mdt_lock_handle *lh, __u64 ibits)
+{
+	return mdt_object_lock_internal(info, o, lh, ibits, false, false);
+}
+
+int mdt_reint_object_lock(struct mdt_thread_info *info, struct mdt_object *o,
+			  struct mdt_lock_handle *lh, __u64 ibits,
+			  bool cos_incompat)
 {
-	return mdt_object_lock_internal(info, o, lh, ibits, false, locality);
+	LASSERT(lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX);
+	return mdt_object_lock_internal(info, o, lh, ibits, false,
+					cos_incompat);
 }
 
 int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o,
-		        struct mdt_lock_handle *lh, __u64 ibits, int locality)
+			struct mdt_lock_handle *lh, __u64 ibits)
+{
+	struct mdt_lock_handle tmp = *lh;
+	int rc;
+
+	rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, false);
+	if (rc == 0)
+		*lh = tmp;
+
+	return rc == 0;
+}
+
+int mdt_reint_object_lock_try(struct mdt_thread_info *info,
+			      struct mdt_object *o, struct mdt_lock_handle *lh,
+			      __u64 ibits, bool cos_incompat)
 {
 	struct mdt_lock_handle tmp = *lh;
 	int rc;
 
-	rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, locality);
+	LASSERT(lh->mlh_reg_mode == LCK_PW || lh->mlh_reg_mode == LCK_EX);
+	rc = mdt_object_lock_internal(info, o, &tmp, ibits, true, cos_incompat);
 	if (rc == 0)
 		*lh = tmp;
 
@@ -2556,38 +2672,40 @@ int mdt_object_lock_try(struct mdt_thread_info *info, struct mdt_object *o,
  * \param mode lock mode
  * \param decref force immediate lock releasing
  */
-static
-void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
-                   ldlm_mode_t mode, int decref)
+static void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
+			  enum ldlm_mode mode, int decref)
 {
-        ENTRY;
+	ENTRY;
 
-        if (lustre_handle_is_used(h)) {
-                if (decref || !info->mti_has_trans ||
-                    !(mode & (LCK_PW | LCK_EX))){
-                        mdt_fid_unlock(h, mode);
-                } else {
-                        struct mdt_device *mdt = info->mti_mdt;
-                        struct ldlm_lock *lock = ldlm_handle2lock(h);
-                        struct ptlrpc_request *req = mdt_info_req(info);
-                        int no_ack = 0;
+	if (lustre_handle_is_used(h)) {
+		if (decref || !info->mti_has_trans ||
+		    !(mode & (LCK_PW | LCK_EX))) {
+			mdt_fid_unlock(h, mode);
+		} else {
+			struct mdt_device *mdt = info->mti_mdt;
+			struct ldlm_lock *lock = ldlm_handle2lock(h);
+			struct ptlrpc_request *req = mdt_info_req(info);
+			int cos;
+
+			cos = (mdt_cos_is_enabled(mdt) ||
+			       mdt_slc_is_enabled(mdt));
+
+			LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n",
+				 h->cookie);
 
-                        LASSERTF(lock != NULL, "no lock for cookie "LPX64"\n",
-                                 h->cookie);
 			/* there is no request if mdt_object_unlock() is called
 			 * from mdt_export_cleanup()->mdt_add_dirty_flag() */
 			if (likely(req != NULL)) {
 				CDEBUG(D_HA, "request = %p reply state = %p"
 				       " transno = "LPD64"\n", req,
 				       req->rq_reply_state, req->rq_transno);
-				if (mdt_cos_is_enabled(mdt)) {
-					no_ack = 1;
+				if (cos) {
 					ldlm_lock_downgrade(lock, LCK_COS);
 					mode = LCK_COS;
 				}
-				ptlrpc_save_lock(req, h, mode, no_ack);
+				ptlrpc_save_lock(req, h, mode, cos);
 			} else {
-				ldlm_lock_decref(h, mode);
+				mdt_fid_unlock(h, mode);
 			}
                         if (mdt_is_lock_sync(lock)) {
                                 CDEBUG(D_HA, "found sync-lock,"
@@ -2604,10 +2722,45 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
 }
 
 /**
+ * Save cross-MDT lock in uncommitted_slc_locks
+ *
+ * Keep the lock referenced until transaction commit happens or release the lock
+ * immediately depending on input parameters.
+ *
+ * \param info thead info object
+ * \param h lock handle
+ * \param mode lock mode
+ * \param decref force immediate lock releasing
+ */
+static void mdt_save_remote_lock(struct mdt_thread_info *info,
+				 struct lustre_handle *h, enum ldlm_mode mode,
+				 int decref)
+{
+	ENTRY;
+
+	if (lustre_handle_is_used(h)) {
+		if (decref || !info->mti_has_trans ||
+		    !(mode & (LCK_PW | LCK_EX))) {
+			ldlm_lock_decref_and_cancel(h, mode);
+		} else {
+			struct ldlm_lock *lock = ldlm_handle2lock(h);
+			struct ptlrpc_request *req = mdt_info_req(info);
+
+			LASSERT(req != NULL);
+			tgt_save_slc_lock(lock, req->rq_transno);
+			ldlm_lock_decref(h, mode);
+		}
+		h->cookie = 0ull;
+	}
+
+	EXIT;
+}
+
+/**
  * Unlock mdt object.
  *
  * Immeditely release the regular lock and the PDO lock or save the
- * lock in reqeuest and keep them referenced until client ACK or
+ * lock in request and keep them referenced until client ACK or
  * transaction commit.
  *
  * \param info thread info object
@@ -2616,17 +2769,15 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
  * \param decref force immediate lock releasing
  */
 void mdt_object_unlock(struct mdt_thread_info *info, struct mdt_object *o,
-                       struct mdt_lock_handle *lh, int decref)
+		       struct mdt_lock_handle *lh, int decref)
 {
-        ENTRY;
-
-        mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
-        mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
+	ENTRY;
 
-	if (lustre_handle_is_used(&lh->mlh_rreg_lh))
-		ldlm_lock_decref(&lh->mlh_rreg_lh, lh->mlh_rreg_mode);
+	mdt_save_lock(info, &lh->mlh_pdo_lh, lh->mlh_pdo_mode, decref);
+	mdt_save_lock(info, &lh->mlh_reg_lh, lh->mlh_reg_mode, decref);
+	mdt_save_remote_lock(info, &lh->mlh_rreg_lh, lh->mlh_rreg_mode, decref);
 
-        EXIT;
+	EXIT;
 }
 
 struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info,
@@ -2640,8 +2791,7 @@ struct mdt_object *mdt_object_find_lock(struct mdt_thread_info *info,
         if (!IS_ERR(o)) {
                 int rc;
 
-                rc = mdt_object_lock(info, o, lh, ibits,
-                                     MDT_LOCAL_LOCK);
+		rc = mdt_object_lock(info, o, lh, ibits);
                 if (rc != 0) {
                         mdt_object_put(info->mti_env, o);
                         o = ERR_PTR(rc);
@@ -2687,35 +2837,19 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags)
         if (body == NULL)
                 RETURN(-EFAULT);
 
-        if (!(body->valid & OBD_MD_FLID))
-                RETURN(0);
+	if (!(body->mbo_valid & OBD_MD_FLID))
+		RETURN(0);
 
-        if (!fid_is_sane(&body->fid1)) {
-                CERROR("Invalid fid: "DFID"\n", PFID(&body->fid1));
+	if (!fid_is_sane(&body->mbo_fid1)) {
+		CERROR("Invalid fid: "DFID"\n", PFID(&body->mbo_fid1));
                 RETURN(-EINVAL);
         }
 
-        /*
-         * Do not get size or any capa fields before we check that request
-         * contains capa actually. There are some requests which do not, for
-         * instance MDS_IS_SUBDIR.
-         */
-        if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) &&
-            req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT))
-                mdt_set_capainfo(info, 0, &body->fid1,
-                                 req_capsule_client_get(pill, &RMF_CAPA1));
-
-        obj = mdt_object_find(env, info->mti_mdt, &body->fid1);
-        if (!IS_ERR(obj)) {
-                if ((flags & HABEO_CORPUS) &&
-                    !mdt_object_exists(obj)) {
-                        mdt_object_put(env, obj);
-                        /* for capability renew ENOENT will be handled in
-                         * mdt_renew_capa */
-                        if (body->valid & OBD_MD_FLOSSCAPA)
-                                rc = 0;
-                        else
-                                rc = -ENOENT;
+	obj = mdt_object_find(env, info->mti_mdt, &body->mbo_fid1);
+	if (!IS_ERR(obj)) {
+		if ((flags & HABEO_CORPUS) && !mdt_object_exists(obj)) {
+			mdt_object_put(env, obj);
+			rc = -ENOENT;
                 } else {
                         info->mti_object = obj;
                         rc = 0;
@@ -2751,17 +2885,6 @@ static int mdt_unpack_req_pack_rep(struct mdt_thread_info *info, __u32 flags)
         RETURN(rc);
 }
 
-static int mdt_init_capa_ctxt(const struct lu_env *env, struct mdt_device *m)
-{
-	struct md_device *next = m->mdt_child;
-
-	return next->md_ops->mdo_init_capa_ctxt(env, next,
-						m->mdt_lut.lut_mds_capa,
-						m->mdt_capa_timeout,
-						m->mdt_capa_alg,
-						m->mdt_capa_keys);
-}
-
 void mdt_lock_handle_init(struct mdt_lock_handle *lh)
 {
         lh->mlh_type = MDT_NUL_LOCK;
@@ -2816,6 +2939,8 @@ void mdt_thread_info_init(struct ptlrpc_request *req,
 
         info->mti_spec.no_create = 0;
 	info->mti_spec.sp_rm_entry = 0;
+	info->mti_spec.sp_permitted = 0;
+	info->mti_spec.sp_migrate_close = 0;
 
 	info->mti_spec.u.sp_ea.eadata = NULL;
 	info->mti_spec.u.sp_ea.eadatalen = 0;
@@ -2843,27 +2968,14 @@ void mdt_thread_info_fini(struct mdt_thread_info *info)
 struct mdt_thread_info *tsi2mdt_info(struct tgt_session_info *tsi)
 {
 	struct mdt_thread_info	*mti;
-	struct lustre_capa	*lc;
 
 	mti = mdt_th_info(tsi->tsi_env);
 	LASSERT(mti != NULL);
 
 	mdt_thread_info_init(tgt_ses_req(tsi), mti);
 	if (tsi->tsi_corpus != NULL) {
-		struct req_capsule *pill = tsi->tsi_pill;
-
 		mti->mti_object = mdt_obj(tsi->tsi_corpus);
 		lu_object_get(tsi->tsi_corpus);
-
-		/*
-		 * XXX: must be part of tgt_mdt_body_unpack but moved here
-		 * due to mdt_set_capainfo().
-		 */
-		if (req_capsule_has_field(pill, &RMF_CAPA1, RCL_CLIENT) &&
-		    req_capsule_get_size(pill, &RMF_CAPA1, RCL_CLIENT) > 0) {
-			lc = req_capsule_client_get(pill, &RMF_CAPA1);
-			mdt_set_capainfo(mti, 0, &tsi->tsi_mdt_body->fid1, lc);
-		}
 	}
 	mti->mti_body = tsi->tsi_mdt_body;
 	mti->mti_dlm_req = tsi->tsi_dlm_req;
@@ -2878,6 +2990,13 @@ static int mdt_tgt_connect(struct tgt_session_info *tsi)
 
 	ENTRY;
 
+	if (OBD_FAIL_CHECK(OBD_FAIL_TGT_DELAY_CONDITIONAL) &&
+	    cfs_fail_val ==
+	    tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id) {
+		set_current_state(TASK_UNINTERRUPTIBLE);
+		schedule_timeout(msecs_to_jiffies(3 * MSEC_PER_SEC));
+	}
+
 	rc = tgt_connect(tsi);
 	if (rc != 0)
 		RETURN(rc);
@@ -2999,19 +3118,17 @@ static struct mdt_it_flavor {
 };
 
 static int
-mdt_intent_lock_replace(struct mdt_thread_info *info, struct ldlm_lock **lockp,
-			struct ldlm_lock *new_lock, struct mdt_lock_handle *lh,
+mdt_intent_lock_replace(struct mdt_thread_info *info,
+			struct ldlm_lock **lockp,
+			struct mdt_lock_handle *lh,
 			__u64 flags)
 {
         struct ptlrpc_request  *req = mdt_info_req(info);
         struct ldlm_lock       *lock = *lockp;
+	struct ldlm_lock       *new_lock;
 
-        /*
-         * Get new lock only for cases when possible resent did not find any
-         * lock.
-         */
-        if (new_lock == NULL)
-                new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0);
+	/* If possible resent found a lock, @lh is set to its handle */
+	new_lock = ldlm_handle2lock_long(&lh->mlh_reg_lh, 0);
 
         if (new_lock == NULL && (flags & LDLM_FL_INTENT_ONLY)) {
                 lh->mlh_reg_lh.cookie = 0;
@@ -3043,6 +3160,8 @@ mdt_intent_lock_replace(struct mdt_thread_info *info, struct ldlm_lock **lockp,
                  */
                 LASSERT(lustre_msg_get_flags(req->rq_reqmsg) &
                         MSG_RESENT);
+
+		LDLM_LOCK_RELEASE(new_lock);
                 lh->mlh_reg_lh.cookie = 0;
                 RETURN(ELDLM_LOCK_REPLACED);
         }
@@ -3084,58 +3203,36 @@ mdt_intent_lock_replace(struct mdt_thread_info *info, struct ldlm_lock **lockp,
 
 static void mdt_intent_fixup_resent(struct mdt_thread_info *info,
 				    struct ldlm_lock *new_lock,
-				    struct ldlm_lock **old_lock,
 				    struct mdt_lock_handle *lh,
-				    enum mdt_it_code opcode)
+				    __u64 flags)
 {
         struct ptlrpc_request  *req = mdt_info_req(info);
-        struct obd_export      *exp = req->rq_export;
-        struct lustre_handle    remote_hdl;
         struct ldlm_request    *dlmreq;
-        struct ldlm_lock       *lock;
 
         if (!(lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT))
                 return;
 
         dlmreq = req_capsule_client_get(info->mti_pill, &RMF_DLM_REQ);
-        remote_hdl = dlmreq->lock_handle[0];
-	/* If the client does not require open lock, it does not need to
-	 * search lock in exp_lock_hash, since the server thread will
-	 * make sure the lock will be released, and the resend request
-	 * can always re-enqueue the lock */
-	if ((opcode != MDT_IT_OPEN) || (opcode == MDT_IT_OPEN &&
-	    info->mti_spec.sp_cr_flags & MDS_OPEN_LOCK)) {
-		/* In the function below, .hs_keycmp resolves to
-		 * ldlm_export_lock_keycmp() */
-		/* coverity[overrun-buffer-val] */
-		lock = cfs_hash_lookup(exp->exp_lock_hash, &remote_hdl);
-		if (lock) {
-			lock_res_and_lock(lock);
-			if (lock != new_lock) {
-				lh->mlh_reg_lh.cookie = lock->l_handle.h_cookie;
-				lh->mlh_reg_mode = lock->l_granted_mode;
-
-				LDLM_DEBUG(lock, "Restoring lock cookie");
-				DEBUG_REQ(D_DLMTRACE, req,
-					  "restoring lock cookie "LPX64,
-					  lh->mlh_reg_lh.cookie);
-				if (old_lock)
-					*old_lock = LDLM_LOCK_GET(lock);
-				cfs_hash_put(exp->exp_lock_hash,
-					     &lock->l_exp_hash);
-				unlock_res_and_lock(lock);
-				return;
-			}
-			cfs_hash_put(exp->exp_lock_hash, &lock->l_exp_hash);
-			unlock_res_and_lock(lock);
-		}
+
+	/* Check if this is a resend case (MSG_RESENT is set on RPC) and a
+	 * lock was found by ldlm_handle_enqueue(); if so @lh must be
+	 * initialized. */
+	if (flags & LDLM_FL_RESENT) {
+		lh->mlh_reg_lh.cookie = new_lock->l_handle.h_cookie;
+		lh->mlh_reg_mode = new_lock->l_granted_mode;
+
+		LDLM_DEBUG(new_lock, "Restoring lock cookie");
+		DEBUG_REQ(D_DLMTRACE, req, "restoring lock cookie "LPX64,
+			  lh->mlh_reg_lh.cookie);
+		return;
 	}
-        /*
-         * If the xid matches, then we know this is a resent request, and allow
-         * it. (It's probably an OPEN, for which we don't send a lock.
-         */
-        if (req_xid_is_last(req))
-                return;
+
+	/*
+	 * If the xid matches, then we know this is a resent request, and allow
+	 * it. (It's probably an OPEN, for which we don't send a lock.
+	 */
+	if (req_can_reconstruct(req, NULL))
+		return;
 
         /*
          * This remote handle isn't enqueued, so we never received or processed
@@ -3144,8 +3241,8 @@ static void mdt_intent_fixup_resent(struct mdt_thread_info *info,
          */
         lustre_msg_clear_flags(req->rq_reqmsg, MSG_RESENT);
 
-        DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64,
-                  remote_hdl.cookie);
+	DEBUG_REQ(D_DLMTRACE, req, "no existing lock with rhandle "LPX64,
+		  dlmreq->lock_handle[0].cookie);
 }
 
 static int mdt_intent_getxattr(enum mdt_it_code opcode,
@@ -3162,19 +3259,18 @@ static int mdt_intent_getxattr(enum mdt_it_code opcode,
 	 * (for the resend case) or a new lock. Below we will use it to
 	 * replace the original lock.
 	 */
-	mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode);
+	mdt_intent_fixup_resent(info, *lockp, lhc, flags);
 	if (!lustre_handle_is_used(&lhc->mlh_reg_lh)) {
 		mdt_lock_reg_init(lhc, (*lockp)->l_req_mode);
 		rc = mdt_object_lock(info, info->mti_object, lhc,
-					MDS_INODELOCK_XATTR,
-					MDT_LOCAL_LOCK);
+				     MDS_INODELOCK_XATTR);
 		if (rc)
 			return rc;
 	}
 
 	grc = mdt_getxattr(info);
 
-	rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
+	rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
 
 	if (mdt_info_req(info)->rq_repmsg != NULL)
 		ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
@@ -3192,7 +3288,6 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
 			      __u64 flags)
 {
         struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_RMT];
-        struct ldlm_lock       *new_lock = NULL;
         __u64                   child_bits;
         struct ldlm_reply      *ldlm_rep;
         struct mdt_body        *reqbody;
@@ -3206,9 +3301,9 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
         LASSERT(repbody);
 
-        info->mti_cross_ref = !!(reqbody->valid & OBD_MD_FLCROSSREF);
-        repbody->eadatasize = 0;
-        repbody->aclsize = 0;
+	info->mti_cross_ref = !!(reqbody->mbo_valid & OBD_MD_FLCROSSREF);
+	repbody->mbo_eadatasize = 0;
+	repbody->mbo_aclsize = 0;
 
         switch (opcode) {
         case MDT_IT_LOOKUP:
@@ -3223,15 +3318,15 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
                 GOTO(out_shrink, rc = -EINVAL);
         }
 
-        rc = mdt_init_ucred(info, reqbody);
-        if (rc)
-                GOTO(out_shrink, rc);
+	rc = mdt_init_ucred_intent_getattr(info, reqbody);
+	if (rc)
+		GOTO(out_shrink, rc);
 
         ldlm_rep = req_capsule_server_get(info->mti_pill, &RMF_DLM_REP);
         mdt_set_disposition(info, ldlm_rep, DISP_IT_EXECD);
 
 	/* Get lock from request for possible resent case. */
-	mdt_intent_fixup_resent(info, *lockp, &new_lock, lhc, opcode);
+	mdt_intent_fixup_resent(info, *lockp, lhc, flags);
 
 	rc = mdt_getattr_name_lock(info, lhc, child_bits, ldlm_rep);
 	ldlm_rep->lock_policy_res2 = clear_serious(rc);
@@ -3244,7 +3339,7 @@ static int mdt_intent_getattr(enum mdt_it_code opcode,
                 GOTO(out_ucred, rc = ELDLM_LOCK_ABORTED);
         }
 
-        rc = mdt_intent_lock_replace(info, lockp, new_lock, lhc, flags);
+	rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
         EXIT;
 out_ucred:
         mdt_exit_ucred(info);
@@ -3261,9 +3356,11 @@ static int mdt_intent_layout(enum mdt_it_code opcode,
 			     struct ldlm_lock **lockp,
 			     __u64 flags)
 {
+	struct mdt_lock_handle *lhc = &info->mti_lh[MDT_LH_LAYOUT];
 	struct layout_intent *layout;
 	struct lu_fid *fid;
 	struct mdt_object *obj = NULL;
+	int layout_size = 0;
 	int rc = 0;
 	ENTRY;
 
@@ -3273,40 +3370,51 @@ static int mdt_intent_layout(enum mdt_it_code opcode,
 		RETURN(-EINVAL);
 	}
 
+	layout = req_capsule_client_get(info->mti_pill, &RMF_LAYOUT_INTENT);
+	if (layout == NULL)
+		RETURN(-EPROTO);
+
+	if (layout->li_opc != LAYOUT_INTENT_ACCESS) {
+		CERROR("%s: Unsupported layout intent opc %d\n",
+		       mdt_obd_name(info->mti_mdt), layout->li_opc);
+		RETURN(-EINVAL);
+	}
+
 	fid = &info->mti_tmp_fid2;
 	fid_extract_from_res_name(fid, &(*lockp)->l_resource->lr_name);
 
+	/* Get lock from request for possible resent case. */
+	mdt_intent_fixup_resent(info, *lockp, lhc, flags);
+
 	obj = mdt_object_find(info->mti_env, info->mti_mdt, fid);
 	if (IS_ERR(obj))
-		RETURN(PTR_ERR(obj));
+		GOTO(out, rc = PTR_ERR(obj));
 
 	if (mdt_object_exists(obj) && !mdt_object_remote(obj)) {
-		/* get the length of lsm */
-		rc = mdt_attr_get_eabuf_size(info, obj);
-		if (rc < 0)
-			RETURN(rc);
+		layout_size = mdt_attr_get_eabuf_size(info, obj);
+		if (layout_size < 0)
+			GOTO(out_obj, rc = layout_size);
 
-		if (rc > info->mti_mdt->mdt_max_mdsize)
-			info->mti_mdt->mdt_max_mdsize = rc;
+		if (layout_size > info->mti_mdt->mdt_max_mdsize)
+			info->mti_mdt->mdt_max_mdsize = layout_size;
 	}
 
-	mdt_object_put(info->mti_env, obj);
-
 	(*lockp)->l_lvb_type = LVB_T_LAYOUT;
-	req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER, rc);
+	req_capsule_set_size(info->mti_pill, &RMF_DLM_LVB, RCL_SERVER,
+			     layout_size);
 	rc = req_capsule_server_pack(info->mti_pill);
-	if (rc != 0)
-		RETURN(-EINVAL);
+	GOTO(out_obj, rc);
 
-	layout = req_capsule_client_get(info->mti_pill, &RMF_LAYOUT_INTENT);
-	LASSERT(layout != NULL);
-	if (layout->li_opc == LAYOUT_INTENT_ACCESS)
-		/* return to normal ldlm handling */
-		RETURN(0);
+out_obj:
+	mdt_object_put(info->mti_env, obj);
+
+	if (rc == 0 && lustre_handle_is_used(&lhc->mlh_reg_lh))
+		rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
+
+out:
+	lhc->mlh_reg_lh.cookie = 0;
 
-	CERROR("%s: Unsupported layout intent (%d)\n",
-		mdt_obd_name(info->mti_mdt), layout->li_opc);
-	RETURN(-EINVAL);
+	return rc;
 }
 
 static int mdt_intent_reint(enum mdt_it_code opcode,
@@ -3337,7 +3445,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
         }
 
 	/* Get lock from request for possible resent case. */
-	mdt_intent_fixup_resent(info, *lockp, NULL, lhc, opcode);
+	mdt_intent_fixup_resent(info, *lockp, lhc, flags);
 
         rc = mdt_reint_internal(info, lhc, opc);
 
@@ -3353,19 +3461,21 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
 
 	/* the open lock or the lock for cross-ref object should be
 	 * returned to the client */
-	if (rc == -EREMOTE || mdt_get_disposition(rep, DISP_OPEN_LOCK)) {
-		LASSERT(lustre_handle_is_used(&lhc->mlh_reg_lh));
+	if (lustre_handle_is_used(&lhc->mlh_reg_lh) &&
+	    (rc == 0 || rc == -MDT_EREMOTE_OPEN)) {
 		rep->lock_policy_res2 = 0;
-		rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
+		rc = mdt_intent_lock_replace(info, lockp, lhc, flags);
 		RETURN(rc);
 	}
 
 	rep->lock_policy_res2 = clear_serious(rc);
 
         if (rep->lock_policy_res2 == -ENOENT &&
-            mdt_get_disposition(rep, DISP_LOOKUP_NEG))
-                rep->lock_policy_res2 = 0;
+	    mdt_get_disposition(rep, DISP_LOOKUP_NEG) &&
+	    !mdt_get_disposition(rep, DISP_OPEN_CREATE))
+		rep->lock_policy_res2 = 0;
 
+	lhc->mlh_reg_lh.cookie = 0ull;
         if (rc == -ENOTCONN || rc == -ENODEV ||
             rc == -EOVERFLOW) { /**< if VBR failure then return error */
                 /*
@@ -3374,91 +3484,76 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
                  * will detect this, then disconnect, reconnect the import
                  * immediately, instead of impacting the following the rpc.
                  */
-                lhc->mlh_reg_lh.cookie = 0ull;
                 RETURN(rc);
-        } else {
-                /*
-                 * For other cases, the error will be returned by intent.
-                 * and client will retrieve the result from intent.
-                 */
-                 /*
-                  * FIXME: when open lock is finished, that should be
-                  * checked here.
-                  */
-                if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
-                        LASSERTF(rc == 0, "Error occurred but lock handle "
-                                 "is still in use, rc = %d\n", rc);
-                        rep->lock_policy_res2 = 0;
-                        rc = mdt_intent_lock_replace(info, lockp, NULL, lhc, flags);
-                        RETURN(rc);
-                } else {
-                        lhc->mlh_reg_lh.cookie = 0ull;
-                        RETURN(ELDLM_LOCK_ABORTED);
-                }
         }
+	/*
+	 * For other cases, the error will be returned by intent, and client
+	 * will retrieve the result from intent.
+	 */
+	RETURN(ELDLM_LOCK_ABORTED);
 }
 
-static int mdt_intent_code(long itcode)
+static int mdt_intent_code(enum ldlm_intent_flags itcode)
 {
-        int rc;
+	int rc;
 
-        switch(itcode) {
-        case IT_OPEN:
-                rc = MDT_IT_OPEN;
-                break;
-        case IT_OPEN|IT_CREAT:
-                rc = MDT_IT_OCREAT;
-                break;
-        case IT_CREAT:
-                rc = MDT_IT_CREATE;
-                break;
-        case IT_READDIR:
-                rc = MDT_IT_READDIR;
-                break;
-        case IT_GETATTR:
-                rc = MDT_IT_GETATTR;
-                break;
-        case IT_LOOKUP:
-                rc = MDT_IT_LOOKUP;
-                break;
-        case IT_UNLINK:
-                rc = MDT_IT_UNLINK;
-                break;
-        case IT_TRUNC:
-                rc = MDT_IT_TRUNC;
-                break;
-        case IT_GETXATTR:
-                rc = MDT_IT_GETXATTR;
-                break;
-        case IT_LAYOUT:
-                rc = MDT_IT_LAYOUT;
-                break;
+	switch (itcode) {
+	case IT_OPEN:
+		rc = MDT_IT_OPEN;
+		break;
+	case IT_OPEN|IT_CREAT:
+		rc = MDT_IT_OCREAT;
+		break;
+	case IT_CREAT:
+		rc = MDT_IT_CREATE;
+		break;
+	case IT_READDIR:
+		rc = MDT_IT_READDIR;
+		break;
+	case IT_GETATTR:
+		rc = MDT_IT_GETATTR;
+		break;
+	case IT_LOOKUP:
+		rc = MDT_IT_LOOKUP;
+		break;
+	case IT_UNLINK:
+		rc = MDT_IT_UNLINK;
+		break;
+	case IT_TRUNC:
+		rc = MDT_IT_TRUNC;
+		break;
+	case IT_GETXATTR:
+		rc = MDT_IT_GETXATTR;
+		break;
+	case IT_LAYOUT:
+		rc = MDT_IT_LAYOUT;
+		break;
 	case IT_QUOTA_DQACQ:
 	case IT_QUOTA_CONN:
 		rc = MDT_IT_QUOTA;
 		break;
-        default:
-                CERROR("Unknown intent opcode: %ld\n", itcode);
-                rc = -EINVAL;
-                break;
-        }
-        return rc;
+	default:
+		CERROR("Unknown intent opcode: 0x%08x\n", itcode);
+		rc = -EINVAL;
+		break;
+	}
+	return rc;
 }
 
-static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
+static int mdt_intent_opc(enum ldlm_intent_flags itopc,
+			  struct mdt_thread_info *info,
 			  struct ldlm_lock **lockp, __u64 flags)
 {
-        struct req_capsule   *pill;
-        struct mdt_it_flavor *flv;
-        int opc;
-        int rc;
-        ENTRY;
-
-        opc = mdt_intent_code(itopc);
-        if (opc < 0)
-                RETURN(-EINVAL);
+	struct req_capsule	*pill = info->mti_pill;
+	struct ptlrpc_request	*req = mdt_info_req(info);
+	struct mdt_it_flavor	*flv;
+	int opc;
+	int rc;
+	ENTRY;
 
-        pill = info->mti_pill;
+	opc = mdt_intent_code(itopc);
+	if (opc < 0)
+		RETURN(-EINVAL);
 
 	if (opc == MDT_IT_QUOTA) {
 		struct lu_device *qmt = info->mti_mdt->mdt_qmt_dev;
@@ -3474,39 +3569,41 @@ static int mdt_intent_opc(long itopc, struct mdt_thread_info *info,
 		RETURN(rc);
 	}
 
-	flv  = &mdt_it_flavor[opc];
-        if (flv->it_fmt != NULL)
-                req_capsule_extend(pill, flv->it_fmt);
+	flv = &mdt_it_flavor[opc];
+	if (flv->it_fmt != NULL)
+		req_capsule_extend(pill, flv->it_fmt);
 
-        rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
-        if (rc == 0) {
-                struct ptlrpc_request *req = mdt_info_req(info);
-		if (flv->it_flags & MUTABOR &&
-		    exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY)
-			RETURN(-EROFS);
-        }
-        if (rc == 0 && flv->it_act != NULL) {
+	rc = mdt_unpack_req_pack_rep(info, flv->it_flags);
+	if (rc < 0)
+		RETURN(rc);
+
+	if (flv->it_flags & MUTABOR &&
+	    exp_connect_flags(req->rq_export) & OBD_CONNECT_RDONLY)
+		RETURN(-EROFS);
+
+	if (flv->it_act != NULL) {
 		struct ldlm_reply *rep;
 
+		OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_INTENT_DELAY, 10);
+
 		/* execute policy */
 		rc = flv->it_act(opc, info, lockp, flags);
 
 		/* Check whether the reply has been packed successfully. */
-		if (mdt_info_req(info)->rq_repmsg != NULL) {
+		if (req->rq_repmsg != NULL) {
 			rep = req_capsule_server_get(info->mti_pill,
 						     &RMF_DLM_REP);
 			rep->lock_policy_res2 =
 				ptlrpc_status_hton(rep->lock_policy_res2);
 		}
-	} else {
-		rc = -EPROTO;
 	}
+
 	RETURN(rc);
 }
 
 static int mdt_intent_policy(struct ldlm_namespace *ns,
-                             struct ldlm_lock **lockp, void *req_cookie,
-			     ldlm_mode_t mode, __u64 flags, void *data)
+			     struct ldlm_lock **lockp, void *req_cookie,
+			     enum ldlm_mode mode, __u64 flags, void *data)
 {
 	struct tgt_session_info	*tsi;
 	struct mdt_thread_info	*info;
@@ -3609,20 +3706,26 @@ static int mdt_register_lwp_callback(void *data)
 
 	LASSERT(mdt_seq_site(mdt)->ss_node_id != 0);
 
-	if (!likely(fld->lsf_new))
-		RETURN(0);
-
 	rc = lu_env_init(&env, LCT_MD_THREAD);
-	if (rc) {
+	if (rc < 0) {
 		CERROR("%s: cannot init env: rc = %d\n", mdt_obd_name(mdt), rc);
 		RETURN(rc);
 	}
 
-	rc = fld_update_from_controller(&env, fld);
-	if (rc != 0) {
-		CERROR("%s: cannot update controller: rc = %d\n",
-		       mdt_obd_name(mdt), rc);
+	/* Allocate new sequence now to avoid creating local transaction
+	 * in the normal transaction process */
+	rc = seq_server_check_and_alloc_super(&env,
+					      mdt_seq_site(mdt)->ss_server_seq);
+	if (rc < 0)
 		GOTO(out, rc);
+
+	if (fld->lsf_new) {
+		rc = fld_update_from_controller(&env, fld);
+		if (rc != 0) {
+			CERROR("%s: cannot update controller: rc = %d\n",
+			       mdt_obd_name(mdt), rc);
+			GOTO(out, rc);
+		}
 	}
 out:
 	lu_env_fini(&env);
@@ -3743,7 +3846,7 @@ static int mdt_seq_init(const struct lu_env *env, struct mdt_device *mdt)
 		GOTO(out_seq_fini, rc);
 
 	if (ss->ss_node_id != 0)
-		/* register controler export through lwp */
+		/* register controller export through lwp */
 		rc = mdt_register_seq_exp(mdt);
 
 	EXIT;
@@ -3823,10 +3926,9 @@ static void mdt_stack_pre_fini(const struct lu_env *env,
 	lustre_cfg_bufs_reset(bufs, mdt_obd_name(m));
 	lustre_cfg_bufs_set_string(bufs, 1, NULL);
 	lcfg = lustre_cfg_new(LCFG_PRE_CLEANUP, bufs);
-	if (!lcfg) {
-		CERROR("%s:Cannot alloc lcfg!\n", mdt_obd_name(m));
-		return;
-	}
+	if (lcfg == NULL)
+		RETURN_EXIT;
+
 	top->ld_ops->ldo_process_config(env, top, lcfg);
 	lustre_cfg_free(lcfg);
 	EXIT;
@@ -3859,10 +3961,9 @@ static void mdt_stack_fini(const struct lu_env *env,
 		strcat(flags, "A");
 	lustre_cfg_bufs_set_string(bufs, 1, flags);
 	lcfg = lustre_cfg_new(LCFG_CLEANUP, bufs);
-	if (!lcfg) {
-		CERROR("Cannot alloc lcfg!\n");
-		return;
-	}
+	if (lcfg == NULL)
+		RETURN_EXIT;
+
 	LASSERT(top);
 	top->ld_ops->ldo_process_config(env, top, lcfg);
 	lustre_cfg_free(lcfg);
@@ -3986,8 +4087,8 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt,
 	lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt);
 
 	lcfg = lustre_cfg_new(LCFG_ATTACH, bufs);
-	if (!lcfg)
-		GOTO(free_bufs, rc = -ENOMEM);
+	if (lcfg == NULL)
+		GOTO(put_profile, rc = -ENOMEM);
 
 	rc = class_attach(lcfg);
 	if (rc)
@@ -3997,7 +4098,7 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt,
 	if (!obd) {
 		CERROR("Can not find obd %s (%s in config)\n",
 		       MDD_OBD_NAME, lustre_cfg_string(cfg, 0));
-		GOTO(class_detach, rc = -EINVAL);
+		GOTO(lcfg_cleanup, rc = -EINVAL);
 	}
 
 	lustre_cfg_free(lcfg);
@@ -4008,6 +4109,8 @@ static int mdt_stack_init(const struct lu_env *env, struct mdt_device *mdt,
 	lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt);
 
 	lcfg = lustre_cfg_new(LCFG_SETUP, bufs);
+	if (lcfg == NULL)
+		GOTO(class_detach, rc = -ENOMEM);
 
 	rc = class_setup(obd, lcfg);
 	if (rc)
@@ -4045,6 +4148,8 @@ class_detach:
 		class_detach(obd, lcfg);
 lcfg_cleanup:
 	lustre_cfg_free(lcfg);
+put_profile:
+	class_put_profile(lprof);
 free_bufs:
 	OBD_FREE_PTR(bufs);
 cleanup_mem:
@@ -4114,8 +4219,8 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt,
 	lustre_cfg_bufs_set_string(bufs, 3, lprof->lp_dt);
 
 	lcfg = lustre_cfg_new(LCFG_ATTACH, bufs);
-	if (!lcfg)
-		GOTO(cleanup_mem, rc = -ENOMEM);
+	if (lcfg == NULL)
+		GOTO(put_profile, rc = -ENOMEM);
 
 	rc = class_attach(lcfg);
 	if (rc)
@@ -4125,7 +4230,7 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt,
 	if (!obd) {
 		CERROR("Can not find obd %s (%s in config)\n", qmtname,
 		       lustre_cfg_string(cfg, 0));
-		GOTO(class_detach, rc = -EINVAL);
+		GOTO(lcfg_cleanup, rc = -EINVAL);
 	}
 
 	lustre_cfg_free(lcfg);
@@ -4139,6 +4244,8 @@ static int mdt_quota_init(const struct lu_env *env, struct mdt_device *mdt,
 				   mdt->mdt_bottom->dd_lu_dev.ld_obd->obd_name);
 
 	lcfg = lustre_cfg_new(LCFG_SETUP, bufs);
+	if (lcfg == NULL)
+		GOTO(class_detach, rc = -ENOMEM);
 
 	rc = class_setup(obd, lcfg);
 	if (rc)
@@ -4175,6 +4282,8 @@ class_detach:
 		class_detach(obd, lcfg);
 lcfg_cleanup:
 	lustre_cfg_free(lcfg);
+put_profile:
+	class_put_profile(lprof);
 cleanup_mem:
 	if (bufs)
 		OBD_FREE_PTR(bufs);
@@ -4206,7 +4315,7 @@ static void mdt_quota_fini(const struct lu_env *env, struct mdt_device *mdt)
 /* mdt_getxattr() is used from mdt_intent_getxattr(), use this wrapper
  * for now. This will be removed along with converting rest of MDT code
  * to use tgt_session_info */
-int mdt_tgt_getxattr(struct tgt_session_info *tsi)
+static int mdt_tgt_getxattr(struct tgt_session_info *tsi)
 {
 	struct mdt_thread_info	*info = tsi2mdt_info(tsi);
 	int			 rc;
@@ -4228,7 +4337,7 @@ TGT_RPC_HANDLER(MDS_FIRST_OPC,
 		HABEO_REFERO,		MDS_SET_INFO,	mdt_set_info,
 		&RQF_OBD_SET_INFO, LUSTRE_MDS_VERSION),
 TGT_MDT_HDL(0,				MDS_GET_INFO,	mdt_get_info),
-TGT_MDT_HDL(0		| HABEO_REFERO,	MDS_GETSTATUS,	mdt_getstatus),
+TGT_MDT_HDL(0		| HABEO_REFERO,	MDS_GET_ROOT,	mdt_get_root),
 TGT_MDT_HDL(HABEO_CORPUS,		MDS_GETATTR,	mdt_getattr),
 TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO,	MDS_GETATTR_NAME,
 							mdt_getattr_name),
@@ -4236,11 +4345,8 @@ TGT_MDT_HDL(HABEO_CORPUS,		MDS_GETXATTR,	mdt_tgt_getxattr),
 TGT_MDT_HDL(0		| HABEO_REFERO,	MDS_STATFS,	mdt_statfs),
 TGT_MDT_HDL(0		| MUTABOR,	MDS_REINT,	mdt_reint),
 TGT_MDT_HDL(HABEO_CORPUS,		MDS_CLOSE,	mdt_close),
-TGT_MDT_HDL(HABEO_CORPUS,		MDS_DONE_WRITING,
-							mdt_done_writing),
 TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO,	MDS_READPAGE,	mdt_readpage),
 TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO,	MDS_SYNC,	mdt_sync),
-TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO,	MDS_IS_SUBDIR,	mdt_is_subdir),
 TGT_MDT_HDL(0,				MDS_QUOTACTL,	mdt_quotactl),
 TGT_MDT_HDL(HABEO_CORPUS| HABEO_REFERO | MUTABOR, MDS_HSM_PROGRESS,
 							mdt_hsm_progress),
@@ -4339,9 +4445,9 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
 	stop.ls_flags = 0;
 	next->md_ops->mdo_iocontrol(env, next, OBD_IOC_STOP_LFSCK, 0, &stop);
 
+	mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child));
 	target_recovery_fini(obd);
 	ping_evictor_stop();
-	mdt_stack_pre_fini(env, m, md2lu_dev(m->mdt_child));
 
 	if (m->mdt_opts.mo_coordinator)
 		mdt_hsm_cdt_stop(m);
@@ -4350,6 +4456,11 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
 
 	mdt_llog_ctxt_unclone(env, m, LLOG_AGENT_ORIG_CTXT);
         mdt_llog_ctxt_unclone(env, m, LLOG_CHANGELOG_ORIG_CTXT);
+
+	if (m->mdt_namespace != NULL)
+		ldlm_namespace_free_prior(m->mdt_namespace, NULL,
+					  d->ld_obd->obd_force);
+
         obd_exports_barrier(obd);
         obd_zombie_barrier();
 
@@ -4360,28 +4471,18 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
         upcall_cache_cleanup(m->mdt_identity_cache);
         m->mdt_identity_cache = NULL;
 
-        if (m->mdt_namespace != NULL) {
-                ldlm_namespace_free(m->mdt_namespace, NULL,
-                                    d->ld_obd->obd_force);
-                d->ld_obd->obd_namespace = m->mdt_namespace = NULL;
-        }
+	if (m->mdt_namespace != NULL) {
+		ldlm_namespace_free_post(m->mdt_namespace);
+		d->ld_obd->obd_namespace = m->mdt_namespace = NULL;
+	}
 
 	mdt_quota_fini(env, m);
 
-        cfs_free_nidlist(&m->mdt_nosquash_nids);
-        if (m->mdt_nosquash_str) {
-                OBD_FREE(m->mdt_nosquash_str, m->mdt_nosquash_strlen);
-                m->mdt_nosquash_str = NULL;
-                m->mdt_nosquash_strlen = 0;
-        }
+	cfs_free_nidlist(&m->mdt_squash.rsi_nosquash_nids);
 
         mdt_seq_fini(env, m);
         mdt_fld_fini(env, m);
 
-        next->md_ops->mdo_init_capa_ctxt(env, next, 0, 0, 0, NULL);
-        cfs_timer_disarm(&m->mdt_ck_timer);
-        mdt_ck_thread_stop(m);
-
 	/*
 	 * Finish the stack
 	 */
@@ -4434,9 +4535,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
         LASSERT(obd != NULL);
 
         m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */
-
-        m->mdt_som_conf = 0;
-
+	m->mdt_opts.mo_evict_tgt_nids = 1;
         m->mdt_opts.mo_cos = MDT_COS_DEFAULT;
 
 	/* default is coordinator off, it is started through conf_param
@@ -4453,23 +4552,23 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
                 LASSERT(num);
                 node_id = simple_strtol(num, NULL, 10);
 		obd->u.obt.obt_magic = OBT_MAGIC;
+		if (lsi->lsi_lmd != NULL &&
+		    lsi->lsi_lmd->lmd_flags & LMD_FLG_SKIP_LFSCK)
+			m->mdt_skip_lfsck = 1;
 	}
 
-	spin_lock_init(&m->mdt_ioepoch_lock);
-        m->mdt_capa_timeout = CAPA_TIMEOUT;
-        m->mdt_capa_alg = CAPA_HMAC_ALG_SHA1;
-        m->mdt_ck_timeout = CAPA_KEY_TIMEOUT;
-        m->mdt_squash_uid = 0;
-        m->mdt_squash_gid = 0;
-        CFS_INIT_LIST_HEAD(&m->mdt_nosquash_nids);
-        m->mdt_nosquash_str = NULL;
-        m->mdt_nosquash_strlen = 0;
-	init_rwsem(&m->mdt_squash_sem);
+	m->mdt_squash.rsi_uid = 0;
+	m->mdt_squash.rsi_gid = 0;
+	INIT_LIST_HEAD(&m->mdt_squash.rsi_nosquash_nids);
+	init_rwsem(&m->mdt_squash.rsi_sem);
 	spin_lock_init(&m->mdt_osfs_lock);
 	m->mdt_osfs_age = cfs_time_shift_64(-1000);
 	m->mdt_enable_remote_dir = 0;
 	m->mdt_enable_remote_dir_gid = 0;
 
+	atomic_set(&m->mdt_mds_mds_conns, 0);
+	atomic_set(&m->mdt_async_commit_count, 0);
+
 	m->mdt_lu_dev.ld_ops = &mdt_lu_ops;
 	m->mdt_lu_dev.ld_obd = obd;
 	/* Set this lu_device to obd for error handling purposes. */
@@ -4531,8 +4630,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
         /* set obd_namespace for compatibility with old code */
         obd->obd_namespace = m->mdt_namespace;
 
-        cfs_timer_init(&m->mdt_ck_timer, mdt_ck_timer_callback, m);
-
 	rc = mdt_hsm_cdt_init(m);
 	if (rc != 0) {
 		CERROR("%s: error initializing coordinator, rc %d\n",
@@ -4540,15 +4637,11 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
                 GOTO(err_free_ns, rc);
 	}
 
-        rc = mdt_ck_thread_start(m);
-        if (rc)
-                GOTO(err_free_hsm, rc);
-
 	rc = tgt_init(env, &m->mdt_lut, obd, m->mdt_bottom, mdt_common_slice,
 		      OBD_FAIL_MDS_ALL_REQUEST_NET,
 		      OBD_FAIL_MDS_ALL_REPLY_NET);
 	if (rc)
-		GOTO(err_capa, rc);
+		GOTO(err_free_hsm, rc);
 
 	rc = mdt_fs_setup(env, m, obd, lsi);
 	if (rc)
@@ -4610,8 +4703,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
 	 * when the whole stack is complete and ready
 	 * to serve the requests */
 
-        mdt_init_capa_ctxt(env, m);
-
         /* Reduce the initial timeout on an MDS because it doesn't need such
          * a long timeout as an OST does. Adaptive timeouts will adjust this
          * value appropriately. */
@@ -4629,9 +4720,6 @@ err_fs_cleanup:
 	mdt_fs_cleanup(env, m);
 err_tgt:
 	tgt_fini(env, &m->mdt_lut);
-err_capa:
-	cfs_timer_disarm(&m->mdt_ck_timer);
-	mdt_ck_thread_stop(m);
 err_free_hsm:
 	mdt_hsm_cdt_fini(m);
 err_free_ns:
@@ -4656,6 +4744,7 @@ static struct cfg_interop_param mdt_interop_param[] = {
 	{ "mdt.group_upcall",	NULL },
 	{ "mdt.quota_type",	NULL },
 	{ "mdd.quota_type",	NULL },
+	{ "mdt.som",		NULL },
 	{ "mdt.rootsquash",	"mdt.root_squash" },
 	{ "mdt.nosquash_nid",	"mdt.nosquash_nids" },
 	{ NULL }
@@ -4673,7 +4762,6 @@ static int mdt_process_config(const struct lu_env *env,
 
 	switch (cfg->lcfg_command) {
 	case LCFG_PARAM: {
-		struct lprocfs_static_vars  lvars;
 		struct obd_device	   *obd = d->ld_obd;
 
 		/* For interoperability */
@@ -4708,8 +4796,7 @@ static int mdt_process_config(const struct lu_env *env,
 			}
 		}
 
-		lprocfs_mdt_init_vars(&lvars);
-		rc = class_process_proc_param(PARAM_MDT, lvars.obd_vars,
+		rc = class_process_proc_param(PARAM_MDT, obd->obd_vars,
 					      cfg, obd);
 		if (rc > 0 || rc == -ENOSYS) {
 			/* is it an HSM var ? */
@@ -4754,7 +4841,7 @@ static struct lu_object *mdt_object_alloc(const struct lu_env *env,
 		lu_object_init(o, h, d);
 		lu_object_add_top(h, o);
 		o->lo_ops = &mdt_obj_ops;
-		mutex_init(&mo->mot_ioepoch_mutex);
+		spin_lock_init(&mo->mot_write_lock);
 		mutex_init(&mo->mot_lov_mutex);
 		init_rwsem(&mo->mot_open_sem);
 		RETURN(o);
@@ -4805,13 +4892,13 @@ static void mdt_object_free(const struct lu_env *env, struct lu_object *o)
 }
 
 static int mdt_object_print(const struct lu_env *env, void *cookie,
-                            lu_printer_t p, const struct lu_object *o)
+			    lu_printer_t p, const struct lu_object *o)
 {
-        struct mdt_object *mdto = mdt_obj((struct lu_object *)o);
-        return (*p)(env, cookie, LUSTRE_MDT_NAME"-object@%p(ioepoch="LPU64" "
-                    "flags="LPX64", epochcount=%d, writecount=%d)",
-                    mdto, mdto->mot_ioepoch, mdto->mot_flags,
-                    mdto->mot_ioepoch_count, mdto->mot_writecount);
+	struct mdt_object *mdto = mdt_obj((struct lu_object *)o);
+
+	return (*p)(env, cookie,
+		    LUSTRE_MDT_NAME"-object@%p(flags=%d, writecount=%d)",
+		    mdto, mdto->mot_flags, mdto->mot_write_count);
 }
 
 static int mdt_prepare(const struct lu_env *env,
@@ -4821,7 +4908,6 @@ static int mdt_prepare(const struct lu_env *env,
 	struct mdt_device *mdt = mdt_dev(cdev);
 	struct lu_device *next = &mdt->mdt_child->md_lu_dev;
 	struct obd_device *obd = cdev->ld_obd;
-	struct lfsck_start_param lsp;
 	int rc;
 
 	ENTRY;
@@ -4845,17 +4931,6 @@ static int mdt_prepare(const struct lu_env *env,
 	 * register the namespace to such instance. */
 	LASSERTF(rc == 0, "register namespace failed: rc = %d\n", rc);
 
-	lsp.lsp_start = NULL;
-	lsp.lsp_index_valid = 0;
-	rc = mdt->mdt_child->md_ops->mdo_iocontrol(env, mdt->mdt_child,
-						   OBD_IOC_START_LFSCK,
-						   0, &lsp);
-	if (rc != 0) {
-		CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n",
-		      mdt_obd_name(mdt), rc);
-		rc = 0;
-	}
-
 	if (mdt->mdt_seq_site.ss_node_id == 0) {
 		rc = mdt->mdt_child->md_ops->mdo_root_get(env, mdt->mdt_child,
 							 &mdt->mdt_md_root_fid);
@@ -4864,6 +4939,7 @@ static int mdt_prepare(const struct lu_env *env,
 	}
 
 	LASSERT(!test_bit(MDT_FL_CFGLOG, &mdt->mdt_state));
+
 	target_recovery_init(&mdt->mdt_lut, tgt_request_handle);
 	set_bit(MDT_FL_CFGLOG, &mdt->mdt_state);
 	LASSERT(obd->obd_no_conn);
@@ -4937,12 +5013,14 @@ static int mdt_connect_internal(struct obd_export *exp,
 	data->ocd_connect_flags &= MDT_CONNECT_SUPPORTED;
 	data->ocd_ibits_known &= MDS_INODELOCK_FULL;
 
-	/* If no known bits (which should not happen, probably,
-	   as everybody should support LOOKUP and UPDATE bits at least)
-	   revert to compat mode with plain locks. */
-	if (!data->ocd_ibits_known &&
-	    data->ocd_connect_flags & OBD_CONNECT_IBITS)
-		data->ocd_connect_flags &= ~OBD_CONNECT_IBITS;
+	if (!(data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
+	    !(data->ocd_connect_flags & OBD_CONNECT_IBITS)) {
+		CWARN("%s: client %s does not support ibits lock, either "
+		      "very old or an invalid client: flags "LPX64"\n",
+		      mdt_obd_name(mdt), exp->exp_client_uuid.uuid,
+		      data->ocd_connect_flags);
+		return -EBADE;
+	}
 
 	if (!mdt->mdt_opts.mo_acl)
 		data->ocd_connect_flags &= ~OBD_CONNECT_ACL;
@@ -4950,9 +5028,6 @@ static int mdt_connect_internal(struct obd_export *exp,
 	if (!mdt->mdt_opts.mo_user_xattr)
 		data->ocd_connect_flags &= ~OBD_CONNECT_XATTR;
 
-	if (!mdt->mdt_som_conf)
-		data->ocd_connect_flags &= ~OBD_CONNECT_SOM;
-
 	if (data->ocd_connect_flags & OBD_CONNECT_BRW_SIZE) {
 		data->ocd_brw_size = min(data->ocd_brw_size,
 					 (__u32)MD_MAX_BRW_SIZE);
@@ -4990,15 +5065,6 @@ static int mdt_connect_internal(struct obd_export *exp,
 		return -EBADE;
 	}
 
-	if (mdt->mdt_som_conf &&
-	    !(data->ocd_connect_flags & (OBD_CONNECT_LIGHTWEIGHT |
-					 OBD_CONNECT_MDS_MDS |
-					 OBD_CONNECT_SOM))) {
-		CWARN("%s: MDS has SOM enabled, but client does not support "
-		      "it\n", mdt_obd_name(mdt));
-		return -EBADE;
-	}
-
 	if (OCD_HAS_FLAG(data, PINGLESS)) {
 		if (ptlrpc_pinger_suppress_pings()) {
 			spin_lock(&exp->exp_obd->obd_dev_lock);
@@ -5011,91 +5077,19 @@ static int mdt_connect_internal(struct obd_export *exp,
 
 	data->ocd_max_easize = mdt->mdt_max_ea_size;
 
-	return 0;
-}
-
-/* mds_connect copy */
-static int mdt_obd_connect(const struct lu_env *env,
-                           struct obd_export **exp, struct obd_device *obd,
-                           struct obd_uuid *cluuid,
-                           struct obd_connect_data *data,
-                           void *localdata)
-{
-        struct obd_export      *lexp;
-        struct lustre_handle    conn = { 0 };
-        struct mdt_device      *mdt;
-        int                     rc;
-        ENTRY;
-
-        LASSERT(env != NULL);
-        if (!exp || !obd || !cluuid)
-                RETURN(-EINVAL);
-
-	mdt = mdt_dev(obd->obd_lu_dev);
-
-	/*
-	 * first, check whether the stack is ready to handle requests
-	 * XXX: probably not very appropriate method is used now
-	 *      at some point we should find a better one
-	 */
-	if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL &&
-	    !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) {
-		rc = obd_get_info(env, mdt->mdt_child_exp,
-				  sizeof(KEY_OSP_CONNECTED),
-				  KEY_OSP_CONNECTED, NULL, NULL, NULL);
-		if (rc)
-			RETURN(-EAGAIN);
-		set_bit(MDT_FL_SYNCED, &mdt->mdt_state);
+	/* NB: Disregard the rule against updating
+	 * exp_connect_data.ocd_connect_flags in this case, since
+	 * tgt_client_new() needs to know if this is client supports
+	 * multiple modify RPCs, and it is safe to expose this flag before
+	 * connection processing completes. */
+	if (data->ocd_connect_flags & OBD_CONNECT_MULTIMODRPCS) {
+		data->ocd_maxmodrpcs = max_mod_rpcs_per_client;
+		spin_lock(&exp->exp_lock);
+		*exp_connect_flags_ptr(exp) |= OBD_CONNECT_MULTIMODRPCS;
+		spin_unlock(&exp->exp_lock);
 	}
 
-        rc = class_connect(&conn, obd, cluuid);
-        if (rc)
-                RETURN(rc);
-
-        lexp = class_conn2export(&conn);
-        LASSERT(lexp != NULL);
-
-        rc = mdt_connect_internal(lexp, mdt, data);
-        if (rc == 0) {
-                struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd;
-
-                LASSERT(lcd);
-		memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid);
-		rc = tgt_client_new(env, lexp);
-                if (rc == 0)
-                        mdt_export_stats_init(obd, lexp, localdata);
-
-		/* For phase I, sync for cross-ref operation. */
-		lexp->exp_keep_sync = 1;
-        }
-
-        if (rc != 0) {
-                class_disconnect(lexp);
-                *exp = NULL;
-        } else {
-                *exp = lexp;
-        }
-
-        RETURN(rc);
-}
-
-static int mdt_obd_reconnect(const struct lu_env *env,
-                             struct obd_export *exp, struct obd_device *obd,
-                             struct obd_uuid *cluuid,
-                             struct obd_connect_data *data,
-                             void *localdata)
-{
-        int                     rc;
-        ENTRY;
-
-        if (exp == NULL || obd == NULL || cluuid == NULL)
-                RETURN(-EINVAL);
-
-        rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data);
-        if (rc == 0)
-                mdt_export_stats_init(obd, exp, localdata);
-
-        RETURN(rc);
+	return 0;
 }
 
 static int mdt_ctxt_add_dirty_flag(struct lu_env *env,
@@ -5125,25 +5119,26 @@ static int mdt_ctxt_add_dirty_flag(struct lu_env *env,
 
 static int mdt_export_cleanup(struct obd_export *exp)
 {
-        struct mdt_export_data *med = &exp->exp_mdt_data;
-        struct obd_device      *obd = exp->exp_obd;
-        struct mdt_device      *mdt;
-        struct mdt_thread_info *info;
-        struct lu_env           env;
-        CFS_LIST_HEAD(closing_list);
-        struct mdt_file_data *mfd, *n;
-        int rc = 0;
-        ENTRY;
+	struct list_head	 closing_list;
+	struct mdt_export_data	*med = &exp->exp_mdt_data;
+	struct obd_device	*obd = exp->exp_obd;
+	struct mdt_device	*mdt;
+	struct mdt_thread_info	*info;
+	struct lu_env		 env;
+	struct mdt_file_data	*mfd, *n;
+	int rc = 0;
+	ENTRY;
 
+	INIT_LIST_HEAD(&closing_list);
 	spin_lock(&med->med_open_lock);
-	while (!cfs_list_empty(&med->med_open_head)) {
-		cfs_list_t *tmp = med->med_open_head.next;
-		mfd = cfs_list_entry(tmp, struct mdt_file_data, mfd_list);
+	while (!list_empty(&med->med_open_head)) {
+		struct list_head *tmp = med->med_open_head.next;
+		mfd = list_entry(tmp, struct mdt_file_data, mfd_list);
 
 		/* Remove mfd handle so it can't be found again.
 		 * We are consuming the mfd_list reference here. */
 		class_handle_unhash(&mfd->mfd_handle);
-		cfs_list_move_tail(&mfd->mfd_list, &closing_list);
+		list_move_tail(&mfd->mfd_list, &closing_list);
 	}
 	spin_unlock(&med->med_open_lock);
         mdt = mdt_dev(obd->obd_lu_dev);
@@ -5160,12 +5155,13 @@ static int mdt_export_cleanup(struct obd_export *exp)
         info->mti_mdt = mdt;
         info->mti_exp = exp;
 
-        if (!cfs_list_empty(&closing_list)) {
-                struct md_attr *ma = &info->mti_attr;
+	if (!list_empty(&closing_list)) {
+		struct md_attr *ma = &info->mti_attr;
 
-                /* Close any open files (which may also cause orphan unlinking). */
-                cfs_list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
-                        cfs_list_del_init(&mfd->mfd_list);
+		/* Close any open files (which may also cause orphan
+		 * unlinking). */
+		list_for_each_entry_safe(mfd, n, &closing_list, mfd_list) {
+			list_del_init(&mfd->mfd_list);
 			ma->ma_need = ma->ma_valid = 0;
 
 			/* This file is being closed due to an eviction, it
@@ -5182,7 +5178,7 @@ static int mdt_export_cleanup(struct obd_export *exp)
 			 * archive request into a noop if it's not actually
 			 * dirty.
 			 */
-			if (mfd->mfd_mode & (FMODE_WRITE|MDS_FMODE_TRUNC))
+			if (mfd->mfd_mode & FMODE_WRITE)
 				rc = mdt_ctxt_add_dirty_flag(&env, info, mfd);
 
 			/* Don't unlink orphan on failover umount, LU-184 */
@@ -5203,6 +5199,18 @@ static int mdt_export_cleanup(struct obd_export *exp)
         RETURN(rc);
 }
 
+static inline void mdt_enable_slc(struct mdt_device *mdt)
+{
+	if (mdt->mdt_lut.lut_sync_lock_cancel == NEVER_SYNC_ON_CANCEL)
+		mdt->mdt_lut.lut_sync_lock_cancel = BLOCKING_SYNC_ON_CANCEL;
+}
+
+static inline void mdt_disable_slc(struct mdt_device *mdt)
+{
+	if (mdt->mdt_lut.lut_sync_lock_cancel == BLOCKING_SYNC_ON_CANCEL)
+		mdt->mdt_lut.lut_sync_lock_cancel = NEVER_SYNC_ON_CANCEL;
+}
+
 static int mdt_obd_disconnect(struct obd_export *exp)
 {
         int rc;
@@ -5211,23 +5219,140 @@ static int mdt_obd_disconnect(struct obd_export *exp)
         LASSERT(exp);
         class_export_get(exp);
 
-        rc = server_disconnect_export(exp);
-        if (rc != 0)
-                CDEBUG(D_IOCTL, "server disconnect error: %d\n", rc);
+	if ((exp_connect_flags(exp) & OBD_CONNECT_MDS_MDS) &&
+	    !(exp_connect_flags(exp) & OBD_CONNECT_LIGHTWEIGHT)) {
+		struct mdt_device *mdt = mdt_dev(exp->exp_obd->obd_lu_dev);
 
-        rc = mdt_export_cleanup(exp);
-        class_export_put(exp);
-        RETURN(rc);
+		if (atomic_dec_and_test(&mdt->mdt_mds_mds_conns))
+			mdt_disable_slc(mdt);
+	}
+
+	rc = server_disconnect_export(exp);
+	if (rc != 0)
+		CDEBUG(D_IOCTL, "server disconnect error: rc = %d\n", rc);
+
+	rc = mdt_export_cleanup(exp);
+	nodemap_del_member(exp);
+	class_export_put(exp);
+	RETURN(rc);
+}
+
+/* mds_connect copy */
+static int mdt_obd_connect(const struct lu_env *env,
+			   struct obd_export **exp, struct obd_device *obd,
+			   struct obd_uuid *cluuid,
+			   struct obd_connect_data *data,
+			   void *localdata)
+{
+	struct obd_export	*lexp;
+	struct lustre_handle	conn = { 0 };
+	struct mdt_device	*mdt;
+	int			 rc;
+	lnet_nid_t		*client_nid = localdata;
+	ENTRY;
+
+	LASSERT(env != NULL);
+	if (!exp || !obd || !cluuid)
+		RETURN(-EINVAL);
+
+	mdt = mdt_dev(obd->obd_lu_dev);
+
+	if ((data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) &&
+	    !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT)) {
+		atomic_inc(&mdt->mdt_mds_mds_conns);
+		mdt_enable_slc(mdt);
+	}
+
+	/*
+	 * first, check whether the stack is ready to handle requests
+	 * XXX: probably not very appropriate method is used now
+	 *      at some point we should find a better one
+	 */
+	if (!test_bit(MDT_FL_SYNCED, &mdt->mdt_state) && data != NULL &&
+	    !(data->ocd_connect_flags & OBD_CONNECT_LIGHTWEIGHT) &&
+	    !(data->ocd_connect_flags & OBD_CONNECT_MDS_MDS)) {
+		rc = obd_get_info(env, mdt->mdt_child_exp,
+				  sizeof(KEY_OSP_CONNECTED),
+				  KEY_OSP_CONNECTED, NULL, NULL);
+		if (rc)
+			RETURN(-EAGAIN);
+		set_bit(MDT_FL_SYNCED, &mdt->mdt_state);
+	}
+
+	rc = class_connect(&conn, obd, cluuid);
+	if (rc)
+		RETURN(rc);
+
+	lexp = class_conn2export(&conn);
+	LASSERT(lexp != NULL);
+
+	rc = nodemap_add_member(*client_nid, lexp);
+	if (rc != 0 && rc != -EEXIST)
+		GOTO(out, rc);
+
+	rc = mdt_connect_internal(lexp, mdt, data);
+	if (rc == 0) {
+		struct lsd_client_data *lcd = lexp->exp_target_data.ted_lcd;
+
+		LASSERT(lcd);
+		memcpy(lcd->lcd_uuid, cluuid, sizeof lcd->lcd_uuid);
+		rc = tgt_client_new(env, lexp);
+		if (rc == 0)
+			mdt_export_stats_init(obd, lexp, localdata);
+	}
+out:
+	if (rc != 0) {
+		class_disconnect(lexp);
+		nodemap_del_member(lexp);
+		*exp = NULL;
+	} else {
+		*exp = lexp;
+		/* Because we do not want this export to be evicted by pinger,
+		 * let's not add this export to the timed chain list. */
+		if (data->ocd_connect_flags & OBD_CONNECT_MDS_MDS) {
+			spin_lock(&lexp->exp_obd->obd_dev_lock);
+			list_del_init(&lexp->exp_obd_chain_timed);
+			spin_unlock(&lexp->exp_obd->obd_dev_lock);
+		}
+	}
+
+	RETURN(rc);
+}
+
+static int mdt_obd_reconnect(const struct lu_env *env,
+			     struct obd_export *exp, struct obd_device *obd,
+			     struct obd_uuid *cluuid,
+			     struct obd_connect_data *data,
+			     void *localdata)
+{
+	lnet_nid_t	       *client_nid = localdata;
+	int                     rc;
+	ENTRY;
+
+	if (exp == NULL || obd == NULL || cluuid == NULL)
+		RETURN(-EINVAL);
+
+	rc = nodemap_add_member(*client_nid, exp);
+	if (rc != 0 && rc != -EEXIST)
+		RETURN(rc);
+
+	rc = mdt_connect_internal(exp, mdt_dev(obd->obd_lu_dev), data);
+	if (rc == 0)
+		mdt_export_stats_init(obd, exp, localdata);
+	else
+		nodemap_del_member(exp);
+
+	RETURN(rc);
 }
 
 /* FIXME: Can we avoid using these two interfaces? */
 static int mdt_init_export(struct obd_export *exp)
 {
-        struct mdt_export_data *med = &exp->exp_mdt_data;
-        int                     rc;
-        ENTRY;
+	struct mdt_export_data *med = &exp->exp_mdt_data;
+	int			rc;
+	ENTRY;
 
-        CFS_INIT_LIST_HEAD(&med->med_open_head);
+	INIT_LIST_HEAD(&med->med_open_head);
 	spin_lock_init(&med->med_open_lock);
 	mutex_init(&med->med_idmap_mutex);
 	med->med_idmap = NULL;
@@ -5272,34 +5397,15 @@ static int mdt_destroy_export(struct obd_export *exp)
                                      &exp->exp_client_uuid)))
                 RETURN(0);
 
-        ldlm_destroy_export(exp);
-        tgt_client_free(exp);
+	ldlm_destroy_export(exp);
+	tgt_client_free(exp);
 
-        LASSERT(cfs_list_empty(&exp->exp_outstanding_replies));
-        LASSERT(cfs_list_empty(&exp->exp_mdt_data.med_open_head));
+	LASSERT(list_empty(&exp->exp_outstanding_replies));
+	LASSERT(list_empty(&exp->exp_mdt_data.med_open_head));
 
-        RETURN(0);
+	RETURN(0);
 }
 
-/** The maximum depth that fid2path() will search.
- * This is limited only because we want to store the fids for
- * historical path lookup purposes.
- */
-#define MAX_PATH_DEPTH 100
-
-/** mdt_path() lookup structure. */
-struct path_lookup_info {
-	__u64			pli_recno;	/**< history point */
-	__u64			pli_currec;	/**< current record */
-	struct lu_fid		pli_fid;
-	struct lu_fid		pli_fids[MAX_PATH_DEPTH]; /**< path, in fids */
-	struct mdt_object	*pli_mdt_obj;
-	char			*pli_path;	/**< full path */
-	int			pli_pathlen;
-	int			pli_linkno;	/**< which hardlink to follow */
-	int			pli_fidcount;	/**< number of \a pli_fids */
-};
-
 int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj,
 		   struct linkea_data *ldata)
 {
@@ -5331,8 +5437,23 @@ int mdt_links_read(struct mdt_thread_info *info, struct mdt_object *mdt_obj,
 	return linkea_init(ldata);
 }
 
+/**
+ * Given an MDT object, try to look up the full path to the object.
+ * Part of the MDT layer implementation of lfs fid2path.
+ *
+ * \param[in]     info  Per-thread common data shared by MDT level handlers.
+ * \param[in]     obj   Object to do path lookup of
+ * \param[in,out] fp    User-provided struct to store path information
+ * \param[in]     root_fid Root FID of current path should reach
+ *
+ * \retval 0 Lookup successful, path information stored in fp
+ * \retval -EAGAIN Lookup failed, usually because object is being moved
+ * \retval negative errno if there was a problem
+ */
 static int mdt_path_current(struct mdt_thread_info *info,
-			    struct path_lookup_info *pli)
+			    struct mdt_object *obj,
+			    struct getinfo_fid2path *fp,
+			    struct lu_fid *root_fid)
 {
 	struct mdt_device	*mdt = info->mti_mdt;
 	struct mdt_object	*mdt_obj;
@@ -5343,8 +5464,9 @@ static int mdt_path_current(struct mdt_thread_info *info,
 	struct lu_buf		*buf = &info->mti_big_buf;
 	char			*ptr;
 	int			reclen;
-	struct linkea_data	ldata = { 0 };
+	struct linkea_data	ldata = { NULL };
 	int			rc = 0;
+	bool			first = true;
 	ENTRY;
 
 	/* temp buffer for path element, the buffer will be finally freed
@@ -5354,122 +5476,159 @@ static int mdt_path_current(struct mdt_thread_info *info,
 		RETURN(-ENOMEM);
 
 	ldata.ld_buf = buf;
-	ptr = pli->pli_path + pli->pli_pathlen - 1;
+	ptr = fp->gf_u.gf_path + fp->gf_pathlen - 1;
 	*ptr = 0;
 	--ptr;
-	pli->pli_fidcount = 0;
-	pli->pli_fids[0] = *(struct lu_fid *)mdt_object_fid(pli->pli_mdt_obj);
-
-	/* root FID only exists on MDT0, and fid2path should also ends at MDT0,
-	 * so checking root_fid can only happen on MDT0. */
-	while (!lu_fid_eq(&mdt->mdt_md_root_fid,
-			  &pli->pli_fids[pli->pli_fidcount])) {
-		mdt_obj = mdt_object_find(info->mti_env, mdt,
-					  &pli->pli_fids[pli->pli_fidcount]);
+	*tmpfid = fp->gf_fid = *mdt_object_fid(obj);
+
+	while (!lu_fid_eq(root_fid, &fp->gf_fid)) {
+		struct lu_buf		lmv_buf;
+
+		if (!lu_fid_eq(root_fid, &mdt->mdt_md_root_fid) &&
+		    lu_fid_eq(&mdt->mdt_md_root_fid, &fp->gf_fid))
+			GOTO(out, rc = -ENOENT);
+
+		mdt_obj = mdt_object_find(info->mti_env, mdt, tmpfid);
 		if (IS_ERR(mdt_obj))
 			GOTO(out, rc = PTR_ERR(mdt_obj));
-		if (mdt_object_remote(mdt_obj)) {
-			mdt_object_put(info->mti_env, mdt_obj);
-			GOTO(remote_out, rc = -EREMOTE);
-		}
+
 		if (!mdt_object_exists(mdt_obj)) {
 			mdt_object_put(info->mti_env, mdt_obj);
 			GOTO(out, rc = -ENOENT);
 		}
 
+		if (mdt_object_remote(mdt_obj)) {
+			mdt_object_put(info->mti_env, mdt_obj);
+			GOTO(remote_out, rc = -EREMOTE);
+		}
+
 		rc = mdt_links_read(info, mdt_obj, &ldata);
-		mdt_object_put(info->mti_env, mdt_obj);
-		if (rc != 0)
+		if (rc != 0) {
+			mdt_object_put(info->mti_env, mdt_obj);
 			GOTO(out, rc);
+		}
 
 		leh = buf->lb_buf;
 		lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
 		linkea_entry_unpack(lee, &reclen, tmpname, tmpfid);
 		/* If set, use link #linkno for path lookup, otherwise use
 		   link #0.  Only do this for the final path element. */
-		if (pli->pli_fidcount == 0 &&
-		    pli->pli_linkno < leh->leh_reccount) {
+		if (first && fp->gf_linkno < leh->leh_reccount) {
 			int count;
-			for (count = 0; count < pli->pli_linkno; count++) {
+			for (count = 0; count < fp->gf_linkno; count++) {
 				lee = (struct link_ea_entry *)
 				     ((char *)lee + reclen);
 				linkea_entry_unpack(lee, &reclen, tmpname,
 						    tmpfid);
 			}
-			if (pli->pli_linkno < leh->leh_reccount - 1)
+			if (fp->gf_linkno < leh->leh_reccount - 1)
 				/* indicate to user there are more links */
-				pli->pli_linkno++;
+				fp->gf_linkno++;
 		}
 
+		lmv_buf.lb_buf = info->mti_xattr_buf;
+		lmv_buf.lb_len = sizeof(info->mti_xattr_buf);
+		/* Check if it is slave stripes */
+		rc = mo_xattr_get(info->mti_env, mdt_object_child(mdt_obj),
+				  &lmv_buf, XATTR_NAME_LMV);
+		mdt_object_put(info->mti_env, mdt_obj);
+		if (rc > 0) {
+			union lmv_mds_md *lmm = lmv_buf.lb_buf;
+
+			/* For slave stripes, get its master */
+			if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE) {
+				fp->gf_fid = *tmpfid;
+				continue;
+			}
+		} else if (rc < 0 && rc != -ENODATA) {
+			GOTO(out, rc);
+		}
+
+		rc = 0;
+
 		/* Pack the name in the end of the buffer */
 		ptr -= tmpname->ln_namelen;
-		if (ptr - 1 <= pli->pli_path)
+		if (ptr - 1 <= fp->gf_u.gf_path)
 			GOTO(out, rc = -EOVERFLOW);
 		strncpy(ptr, tmpname->ln_name, tmpname->ln_namelen);
 		*(--ptr) = '/';
 
-		/* Store the parent fid for historic lookup */
-		if (++pli->pli_fidcount >= MAX_PATH_DEPTH)
-			GOTO(out, rc = -EOVERFLOW);
-		pli->pli_fids[pli->pli_fidcount] = *tmpfid;
+		/* keep the last resolved fid to the client, so the
+		 * client will build the left path on another MDT for
+		 * remote object */
+		fp->gf_fid = *tmpfid;
+
+		first = false;
 	}
 
 remote_out:
 	ptr++; /* skip leading / */
-	memmove(pli->pli_path, ptr, pli->pli_path + pli->pli_pathlen - ptr);
+	memmove(fp->gf_u.gf_path, ptr,
+		fp->gf_u.gf_path + fp->gf_pathlen - ptr);
 
-	EXIT;
 out:
-	return rc;
+	RETURN(rc);
 }
 
-/* Returns the full path to this fid, as of changelog record recno. */
+/**
+ * Given an MDT object, use mdt_path_current to get the path.
+ * Essentially a wrapper to retry mdt_path_current a set number of times
+ * if -EAGAIN is returned (usually because an object is being moved).
+ *
+ * Part of the MDT layer implementation of lfs fid2path.
+ *
+ * \param[in]     info  Per-thread common data shared by mdt level handlers.
+ * \param[in]     obj   Object to do path lookup of
+ * \param[in,out] fp    User-provided struct for arguments and to store path
+ * 			information
+ *
+ * \retval 0 Lookup successful, path information stored in fp
+ * \retval negative errno if there was a problem
+ */
 static int mdt_path(struct mdt_thread_info *info, struct mdt_object *obj,
-		    char *path, int pathlen, __u64 *recno, int *linkno,
-		    struct lu_fid *fid)
+		    struct getinfo_fid2path *fp, struct lu_fid *root_fid)
 {
 	struct mdt_device	*mdt = info->mti_mdt;
-	struct path_lookup_info	*pli;
 	int			tries = 3;
 	int			rc = -EAGAIN;
 	ENTRY;
 
-	if (pathlen < 3)
+	if (fp->gf_pathlen < 3)
 		RETURN(-EOVERFLOW);
 
-	if (lu_fid_eq(&mdt->mdt_md_root_fid, mdt_object_fid(obj))) {
-		path[0] = '\0';
+	if (root_fid == NULL)
+		root_fid = &mdt->mdt_md_root_fid;
+
+	if (lu_fid_eq(root_fid, mdt_object_fid(obj))) {
+		fp->gf_u.gf_path[0] = '\0';
 		RETURN(0);
 	}
 
-	OBD_ALLOC_PTR(pli);
-	if (pli == NULL)
-		RETURN(-ENOMEM);
-
-	pli->pli_mdt_obj = obj;
-	pli->pli_recno = *recno;
-	pli->pli_path = path;
-	pli->pli_pathlen = pathlen;
-	pli->pli_linkno = *linkno;
-
 	/* Retry multiple times in case file is being moved */
 	while (tries-- && rc == -EAGAIN)
-		rc = mdt_path_current(info, pli);
-
-	/* return the last resolved fids to the client, so the client will
-	 * build the left path on another MDT for remote object */
-	*fid = pli->pli_fids[pli->pli_fidcount];
-
-	*recno = pli->pli_currec;
-	/* Return next link index to caller */
-	*linkno = pli->pli_linkno;
-
-	OBD_FREE_PTR(pli);
+		rc = mdt_path_current(info, obj, fp, root_fid);
 
 	RETURN(rc);
 }
 
+/**
+ * Get the full path of the provided FID, as of changelog record recno.
+ *
+ * This checks sanity and looks up object for user provided FID
+ * before calling the actual path lookup code.
+ *
+ * Part of the MDT layer implementation of lfs fid2path.
+ *
+ * \param[in]     info  Per-thread common data shared by mdt level handlers.
+ * \param[in,out] fp    User-provided struct for arguments and to store path
+ * 			information
+ *
+ * \retval 0 Lookup successful, path information and recno stored in fp
+ * \retval -ENOENT, object does not exist
+ * \retval negative errno if there was a problem
+ */
 static int mdt_fid2path(struct mdt_thread_info *info,
+			struct lu_fid *root_fid,
 			struct getinfo_fid2path *fp)
 {
 	struct mdt_device *mdt = info->mti_mdt;
@@ -5491,10 +5650,11 @@ static int mdt_fid2path(struct mdt_thread_info *info,
 	}
 
 	obj = mdt_object_find(info->mti_env, mdt, &fp->gf_fid);
-	if (obj == NULL || IS_ERR(obj)) {
-		CDEBUG(D_IOCTL, "no object "DFID": %ld\n", PFID(&fp->gf_fid),
-		       PTR_ERR(obj));
-		RETURN(-EINVAL);
+	if (IS_ERR(obj)) {
+		rc = PTR_ERR(obj);
+		CDEBUG(D_IOCTL, "cannot find "DFID": rc = %d\n",
+		       PFID(&fp->gf_fid), rc);
+		RETURN(rc);
 	}
 
 	if (mdt_object_remote(obj))
@@ -5506,26 +5666,27 @@ static int mdt_fid2path(struct mdt_thread_info *info,
 
 	if (rc < 0) {
 		mdt_object_put(info->mti_env, obj);
-		CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n",
+		CDEBUG(D_IOCTL, "nonlocal object "DFID": rc = %d\n",
 		       PFID(&fp->gf_fid), rc);
 		RETURN(rc);
 	}
 
-	rc = mdt_path(info, obj, fp->gf_path, fp->gf_pathlen, &fp->gf_recno,
-		      &fp->gf_linkno, &fp->gf_fid);
+	rc = mdt_path(info, obj, fp, root_fid);
 
 	CDEBUG(D_INFO, "fid "DFID", path %s recno "LPX64" linkno %u\n",
-	       PFID(&fp->gf_fid), fp->gf_path, fp->gf_recno, fp->gf_linkno);
+	       PFID(&fp->gf_fid), fp->gf_u.gf_path,
+	       fp->gf_recno, fp->gf_linkno);
 
 	mdt_object_put(info->mti_env, obj);
 
 	RETURN(rc);
 }
 
-static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key,
+static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key, int keylen,
 			    void *val, int vallen)
 {
 	struct getinfo_fid2path *fpout, *fpin;
+	struct lu_fid *root_fid = NULL;
 	int rc = 0;
 
 	fpin = key + cfs_size_round(sizeof(KEY_FID2PATH));
@@ -5538,7 +5699,18 @@ static int mdt_rpc_fid2path(struct mdt_thread_info *info, void *key,
 	if (fpout->gf_pathlen != vallen - sizeof(*fpin))
 		RETURN(-EINVAL);
 
-	rc = mdt_fid2path(info, fpout);
+	if (keylen >= cfs_size_round(sizeof(KEY_FID2PATH)) + sizeof(*fpin) +
+		      sizeof(struct lu_fid)) {
+		/* client sent its root FID, which is normally fileset FID */
+		root_fid = fpin->gf_u.gf_root_fid;
+		if (ptlrpc_req_need_swab(info->mti_pill->rc_req))
+			lustre_swab_lu_fid(root_fid);
+
+		if (root_fid != NULL && !fid_is_sane(root_fid))
+			RETURN(-EINVAL);
+	}
+
+	rc = mdt_fid2path(info, root_fid, fpout);
 	RETURN(rc);
 }
 
@@ -5554,7 +5726,7 @@ int mdt_get_info(struct tgt_session_info *tsi)
 
 	key = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_KEY);
 	if (key == NULL) {
-		CDEBUG(D_IOCTL, "No GETINFO key");
+		CDEBUG(D_IOCTL, "No GETINFO key\n");
 		RETURN(err_serious(-EFAULT));
 	}
 	keylen = req_capsule_get_size(tsi->tsi_pill, &RMF_GETINFO_KEY,
@@ -5562,7 +5734,8 @@ int mdt_get_info(struct tgt_session_info *tsi)
 
 	vallen = req_capsule_client_get(tsi->tsi_pill, &RMF_GETINFO_VALLEN);
 	if (vallen == NULL) {
-		CDEBUG(D_IOCTL, "Unable to get RMF_GETINFO_VALLEN buffer");
+		CDEBUG(D_IOCTL, "%s: cannot get RMF_GETINFO_VALLEN buffer\n",
+				tgt_name(tsi->tsi_tgt));
 		RETURN(err_serious(-EFAULT));
 	}
 
@@ -5574,14 +5747,15 @@ int mdt_get_info(struct tgt_session_info *tsi)
 
 	valout = req_capsule_server_get(tsi->tsi_pill, &RMF_GETINFO_VAL);
 	if (valout == NULL) {
-		CDEBUG(D_IOCTL, "Unable to get get-info RPC out buffer");
+		CDEBUG(D_IOCTL, "%s: cannot get get-info RPC out buffer\n",
+				tgt_name(tsi->tsi_tgt));
 		RETURN(err_serious(-EFAULT));
 	}
 
 	if (KEY_IS(KEY_FID2PATH)) {
 		struct mdt_thread_info	*info = tsi2mdt_info(tsi);
 
-		rc = mdt_rpc_fid2path(info, key, valout, *vallen);
+		rc = mdt_rpc_fid2path(info, key, keylen, valout, *vallen);
 		mdt_thread_info_fini(info);
 	} else {
 		rc = -EINVAL;
@@ -5662,7 +5836,7 @@ static int mdt_ioc_version_get(struct mdt_thread_info *mti, void *karg)
 
 /* ioctls on obd dev */
 static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
-                         void *karg, void *uarg)
+			 void *karg, void __user *uarg)
 {
         struct lu_env      env;
         struct obd_device *obd = exp->exp_obd;
@@ -5685,6 +5859,7 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
                 break;
 	case OBD_IOC_ABORT_RECOVERY:
 		CERROR("%s: Aborting recovery for device\n", mdt_obd_name(mdt));
+		obd->obd_abort_recovery = 1;
 		target_stop_recovery_thread(obd);
 		rc = 0;
 		break;
@@ -5724,6 +5899,14 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 		rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0, &stop);
 		break;
 	}
+	case OBD_IOC_QUERY_LFSCK: {
+		struct md_device	*next = mdt->mdt_child;
+		struct obd_ioctl_data	*data = karg;
+
+		rc = next->md_ops->mdo_iocontrol(&env, next, cmd, 0,
+						 data->ioc_inlbuf1);
+		break;
+	}
         case OBD_IOC_GET_OBJ_VERSION: {
                 struct mdt_thread_info *mti;
                 mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
@@ -5756,12 +5939,25 @@ static int mdt_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 
 static int mdt_postrecov(const struct lu_env *env, struct mdt_device *mdt)
 {
-        struct lu_device *ld = md2lu_dev(mdt->mdt_child);
-        int rc;
-        ENTRY;
+	struct lu_device *ld = md2lu_dev(mdt->mdt_child);
+	int rc;
+	ENTRY;
 
-        rc = ld->ld_ops->ldo_recovery_complete(env, ld);
-        RETURN(rc);
+	if (!mdt->mdt_skip_lfsck) {
+		struct lfsck_start_param lsp;
+
+		lsp.lsp_start = NULL;
+		lsp.lsp_index_valid = 0;
+		rc = mdt->mdt_child->md_ops->mdo_iocontrol(env, mdt->mdt_child,
+							   OBD_IOC_START_LFSCK,
+							   0, &lsp);
+		if (rc != 0 && rc != -EALREADY)
+			CWARN("%s: auto trigger paused LFSCK failed: rc = %d\n",
+			      mdt_obd_name(mdt), rc);
+	}
+
+	rc = ld->ld_ops->ldo_recovery_complete(env, ld);
+	RETURN(rc);
 }
 
 static int mdt_obd_postrecov(struct obd_device *obd)
@@ -5878,13 +6074,13 @@ void mdt_enable_cos(struct mdt_device *mdt, int val)
 
         mdt->mdt_opts.mo_cos = !!val;
         rc = lu_env_init(&env, LCT_LOCAL);
-        if (unlikely(rc != 0)) {
-                CWARN("lu_env initialization failed with rc = %d,"
-                      "cannot sync\n", rc);
-                return;
-        }
-        mdt_device_sync(&env, mdt);
-        lu_env_fini(&env);
+	if (unlikely(rc != 0)) {
+		CWARN("%s: lu_env initialization failed, cannot "
+		      "sync: rc = %d\n", mdt_obd_name(mdt), rc);
+		return;
+	}
+	mdt_device_sync(&env, mdt);
+	lu_env_fini(&env);
 }
 
 /**
@@ -5912,9 +6108,8 @@ static struct lu_device_type mdt_device_type = {
         .ldt_ctx_tags = LCT_MD_THREAD
 };
 
-static int __init mdt_mod_init(void)
+static int __init mdt_init(void)
 {
-	struct lprocfs_static_vars lvars;
 	int rc;
 
 	CLASSERT(sizeof("0x0123456789ABCDEF:0x01234567:0x01234567") ==
@@ -5929,11 +6124,7 @@ static int __init mdt_mod_init(void)
 	if (rc)
 		GOTO(lu_fini, rc);
 
-	lprocfs_mdt_init_vars(&lvars);
 	rc = class_register_type(&mdt_obd_device_ops, NULL, true, NULL,
-#ifndef HAVE_ONLY_PROCFS_SEQ
-				 lvars.module_vars,
-#endif
 				 LUSTRE_MDT_NAME, &mdt_device_type);
 	if (rc)
 		GOTO(mds_fini, rc);
@@ -5946,15 +6137,17 @@ mds_fini:
 	return rc;
 }
 
-static void __exit mdt_mod_exit(void)
+static void __exit mdt_exit(void)
 {
 	class_unregister_type(LUSTRE_MDT_NAME);
 	mds_mod_exit();
 	lu_kmem_fini(mdt_caches);
 }
 
-MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
+MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
 MODULE_DESCRIPTION("Lustre Metadata Target ("LUSTRE_MDT_NAME")");
+MODULE_VERSION(LUSTRE_VERSION_STRING);
 MODULE_LICENSE("GPL");
 
-cfs_module(mdt, LUSTRE_VERSION_STRING, mdt_mod_init, mdt_mod_exit);
+module_init(mdt_init);
+module_exit(mdt_exit);