Whamcloud - gitweb
LU-3540 lod: update recovery thread
[fs/lustre-release.git] / lustre / mdt / mdt_handler.c
index 409f556..a2cc964 100644 (file)
@@ -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, 2014, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -69,6 +69,7 @@
 #include <lustre_param.h>
 #include <lustre_quota.h>
 #include <lustre_lfsck.h>
+#include <lustre_nodemap.h>
 
 mdl_mode_t mdt_mdl_lock_modes[] = {
         [LCK_MINMODE] = MDL_MINMODE,
@@ -127,33 +128,33 @@ 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)
@@ -283,24 +284,6 @@ static int mdt_getstatus(struct tgt_session_info *tsi)
        repbody->mbo_fid1 = mdt->mdt_md_root_fid;
        repbody->mbo_valid |= OBD_MD_FLID;
 
-       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;
-
-               root = mdt_object_find(info->mti_env, mdt, &repbody->mbo_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->mbo_valid |= OBD_MD_FLMDSCAPA;
-        }
        EXIT;
 out:
        mdt_thread_info_fini(info);
@@ -361,57 +344,125 @@ 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)
+int mdt_pack_acl2body(struct mdt_thread_info *info, struct mdt_body *repbody,
+                     struct mdt_object *o, struct lu_nodemap *nodemap)
 {
-        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;
+       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;
 
-       b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
-       b->mbo_size = ma->ma_som->msd_size;
-       b->mbo_blocks = ma->ma_som->msd_blocks;
+       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;
+
+       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->mbo_atime      = attr->la_atime;
-       b->mbo_mtime      = attr->la_mtime;
-       b->mbo_ctime      = attr->la_ctime;
-       b->mbo_mode       = attr->la_mode;
-       b->mbo_size       = attr->la_size;
-       b->mbo_blocks     = attr->la_blocks;
-       b->mbo_uid      = attr->la_uid;
-       b->mbo_gid      = attr->la_gid;
-       b->mbo_flags      = attr->la_flags;
-       b->mbo_nlink      = attr->la_nlink;
-       b->mbo_rdev       = attr->la_rdev;
-
-       /* XXX should pack the reply body according to lu_valid */
-       b->mbo_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 ;
+       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;
@@ -436,17 +487,6 @@ void mdt_pack_attr2body(struct mdt_thread_info *info, struct mdt_body *b,
                b->mbo_valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
        }
 
-       if (fid) {
-               b->mbo_fid1 = *fid;
-               b->mbo_valid |= OBD_MD_FLID;
-               CDEBUG(D_INODE, DFID": nlink=%d, mode=%o, size="LPU64"\n",
-                               PFID(fid), b->mbo_nlink,
-                      b->mbo_mode, b->mbo_size);
-       }
-
-       if (info)
-               mdt_body_reverse_idmap(info, b);
-
        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);
@@ -484,7 +524,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;
@@ -728,19 +769,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);
@@ -779,25 +807,27 @@ 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.*/
@@ -853,8 +883,6 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
                 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)) {
@@ -972,12 +1000,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
         }
 
        if (reqbody->mbo_valid & OBD_MD_FLMODEASIZE) {
-               repbody->mbo_max_cookiesize = 0;
                repbody->mbo_max_mdsize = info->mti_mdt->mdt_max_mdsize;
                repbody->mbo_valid |= OBD_MD_FLMODEASIZE;
-               CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & "
-                      "MAX_COOKIE to : %d:%d\n", repbody->mbo_max_mdsize,
-                      repbody->mbo_max_cookiesize);
+               CDEBUG(D_INODE, "changing the max MD size to %u\n",
+                      repbody->mbo_max_mdsize);
        }
 
        if (exp_connect_rmtclient(info->mti_exp) &&
@@ -997,49 +1023,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
        }
 #ifdef CONFIG_FS_POSIX_ACL
        else if ((exp_connect_flags(req->rq_export) & OBD_CONNECT_ACL) &&
-                (reqbody->mbo_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->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 {
-                               repbody->mbo_aclsize = rc;
-                               repbody->mbo_valid |= OBD_MD_FLACL;
-                               rc = 0;
-                       }
-               }
-       }
+                (reqbody->mbo_valid & OBD_MD_FLACL))
+               rc = mdt_pack_acl2body(info, repbody, o, nodemap);
 #endif
 
-       if (reqbody->mbo_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->mbo_valid |= OBD_MD_FLMDSCAPA;
-        }
-
 out:
         if (rc == 0)
                mdt_counter_incr(req, LPROC_MDT_GETATTR);
@@ -1047,38 +1034,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->mbo_valid |= OBD_MD_FLOSSCAPA;
-        RETURN(rc);
-}
-
 static int mdt_getattr(struct tgt_session_info *tsi)
 {
        struct mdt_thread_info  *info = tsi2mdt_info(tsi);
@@ -1091,15 +1046,6 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 
         reqbody = req_capsule_client_get(pill, &RMF_MDT_BODY);
         LASSERT(reqbody);
-
-       if (reqbody->mbo_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));
 
@@ -1152,11 +1098,6 @@ static int mdt_getattr(struct tgt_session_info *tsi)
 
        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->mbo_fid1, BYPASS_CAPA);
        rc = mdt_getattr_internal(info, obj, 0);
        if (reqbody->mbo_valid & OBD_MD_FLRMTPERM)
                 mdt_exit_ucred(info);
@@ -1171,6 +1112,32 @@ out:
        return rc;
 }
 
+/**
+ * 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)
+{
+       __u64   o1_flags;
+
+       mutex_lock(&o1->mot_lov_mutex);
+       mutex_lock(&o2->mot_lov_mutex);
+
+       o1_flags = o1->mot_flags;
+       o1->mot_flags = (o1->mot_flags & ~MOF_LOV_CREATED) |
+                       (o2->mot_flags & MOF_LOV_CREATED);
+
+       o2->mot_flags = (o2->mot_flags & ~MOF_LOV_CREATED) |
+                       (o1_flags & MOF_LOV_CREATED);
+
+       mutex_unlock(&o2->mot_lov_mutex);
+       mutex_unlock(&o1->mot_lov_mutex);
+}
+
 static int mdt_swap_layouts(struct tgt_session_info *tsi)
 {
        struct mdt_thread_info  *info;
@@ -1195,17 +1162,7 @@ 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->mbo_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->mbo_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,
@@ -1256,7 +1213,11 @@ static int mdt_swap_layouts(struct tgt_session_info *tsi)
 
        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:
@@ -1269,47 +1230,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 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->mbo_valid != OBD_MD_FLID)
-                RETURN(0);
+       struct lu_fid   *child_fid = &info->mti_tmp_fid1;
+       int              rc;
+       ENTRY;
 
-        LASSERT(!info->mti_cross_ref);
+       LASSERT(!info->mti_cross_ref);
 
-        /* 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);
+       /* 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);
-#endif
-        if (rc == 0) {
-                repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
+               repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
                repbody->mbo_fid1 = *child_fid;
                repbody->mbo_valid = OBD_MD_FLID;
-        }
-        RETURN(1);
+               mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_POS);
+       } else if (rc == -ENOENT) {
+               mdt_set_disposition(info, ldlm_rep, DISP_LOOKUP_NEG);
+       }
+
+       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.
@@ -1323,7 +1273,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;
@@ -1338,7 +1287,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. */
@@ -1351,7 +1301,7 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 
                rc = mdt_check_resent_lock(info, child, lhc);
                if (rc < 0) {
-                       RETURN(-EPROTO);
+                       RETURN(rc);
                } else if (rc > 0) {
                        mdt_lock_handle_init(lhc);
                        mdt_lock_reg_init(lhc, LCK_PR);
@@ -1379,7 +1329,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);
@@ -1427,10 +1376,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);
                }
 
@@ -1447,8 +1401,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
 
                 /* 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);
 
@@ -1478,11 +1432,11 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
        if (unlikely(IS_ERR(child)))
                GOTO(out_parent, rc = PTR_ERR(child));
 
+       OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout * 2);
        rc = mdt_check_resent_lock(info, child, lhc);
        if (rc < 0) {
                GOTO(out_child, rc);
        } else if (rc > 0) {
-                OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RESEND, obd_timeout*2);
                 mdt_lock_handle_init(lhc);
                mdt_lock_reg_init(lhc, LCK_PR);
                try_layout = false;
@@ -1554,14 +1508,8 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
         }
 
         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);
@@ -1573,8 +1521,6 @@ static int mdt_getattr_name_lock(struct mdt_thread_info *info,
                         "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);
@@ -1973,6 +1919,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);
@@ -1984,18 +1931,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);
@@ -2034,6 +1975,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));
@@ -2323,14 +2271,20 @@ int mdt_check_resent_lock(struct mdt_thread_info *info,
                          struct mdt_lock_handle *lhc)
 {
        /* the lock might already be gotten in ldlm_handle_enqueue() */
-       if (lustre_handle_is_used(&lhc->mlh_reg_lh)) {
+       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);
-               LASSERTF(lock != NULL, "Invalid lock handle "LPX64"\n",
-                        lhc->mlh_reg_lh.cookie);
+               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)) {
@@ -2360,8 +2314,6 @@ int mdt_remote_object_lock(struct mdt_thread_info *mti,
 
        LASSERT(mdt_object_remote(o));
 
-       LASSERT(ibits == MDS_INODELOCK_UPDATE);
-
        fid_build_reg_res_name(fid, res_id);
 
        memset(einfo, 0, sizeof(*einfo));
@@ -2610,7 +2562,7 @@ void mdt_save_lock(struct mdt_thread_info *info, struct lustre_handle *h,
  * 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
@@ -2698,27 +2650,11 @@ static int mdt_body_unpack(struct mdt_thread_info *info, __u32 flags)
                 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->mbo_fid1,
-                                req_capsule_client_get(pill, &RMF_CAPA1));
-
        obj = mdt_object_find(env, info->mti_mdt, &body->mbo_fid1);
        if (!IS_ERR(obj)) {
-               if ((flags & HABEO_CORPUS) &&
-                   !mdt_object_exists(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->mbo_valid & OBD_MD_FLOSSCAPA)
-                                rc = 0;
-                        else
-                                rc = -ENOENT;
+                       rc = -ENOENT;
                 } else {
                         info->mti_object = obj;
                         rc = 0;
@@ -2754,17 +2690,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;
@@ -2819,6 +2744,7 @@ 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.u.sp_ea.eadata = NULL;
        info->mti_spec.u.sp_ea.eadatalen = 0;
@@ -2846,28 +2772,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->mbo_fid1,
-                                        lc);
-               }
        }
        mti->mti_body = tsi->tsi_mdt_body;
        mti->mti_dlm_req = tsi->tsi_dlm_req;
@@ -2884,8 +2796,10 @@ static int mdt_tgt_connect(struct tgt_session_info *tsi)
 
        if (OBD_FAIL_CHECK(OBD_FAIL_TGT_DELAY_CONDITIONAL) &&
            cfs_fail_val ==
-           tsi2mdt_info(tsi)->mti_mdt->mdt_seq_site.ss_node_id)
-               schedule_timeout_and_set_state(TASK_UNINTERRUPTIBLE, HZ * 3);
+           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)
@@ -3348,8 +3262,8 @@ 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 == -MDT_EREMOTE_OPEN || 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, lhc, flags);
                RETURN(rc);
@@ -3361,6 +3275,7 @@ static int mdt_intent_reint(enum mdt_it_code opcode,
             mdt_get_disposition(rep, DISP_LOOKUP_NEG))
                 rep->lock_policy_res2 = 0;
 
+       lhc->mlh_reg_lh.cookie = 0ull;
         if (rc == -ENOTCONN || rc == -ENODEV ||
             rc == -EOVERFLOW) { /**< if VBR failure then return error */
                 /*
@@ -3369,28 +3284,13 @@ 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, 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)
@@ -3443,17 +3343,16 @@ static int mdt_intent_code(long itcode)
 static int mdt_intent_opc(long 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;
@@ -3469,33 +3368,33 @@ 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;
 
                /* 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);
 }
 
@@ -3604,20 +3503,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);
@@ -3738,7 +3643,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;
@@ -3818,10 +3723,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;
@@ -3854,10 +3758,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);
@@ -3981,7 +3884,7 @@ 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)
+       if (lcfg == NULL)
                GOTO(free_bufs, rc = -ENOMEM);
 
        rc = class_attach(lcfg);
@@ -4003,6 +3906,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)
@@ -4109,7 +4014,7 @@ 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)
+       if (lcfg == NULL)
                GOTO(cleanup_mem, rc = -ENOMEM);
 
        rc = class_attach(lcfg);
@@ -4134,6 +4039,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)
@@ -4201,7 +4108,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;
@@ -4231,8 +4138,6 @@ 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(0,                         MDS_QUOTACTL,   mdt_quotactl),
@@ -4367,10 +4272,6 @@ static void mdt_fini(const struct lu_env *env, struct mdt_device *m)
         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
         */
@@ -4424,8 +4325,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
 
         m->mdt_max_mdsize = MAX_MD_SIZE; /* 4 stripes */
 
-        m->mdt_som_conf = 0;
-
         m->mdt_opts.mo_cos = MDT_COS_DEFAULT;
 
        /* default is coordinator off, it is started through conf_param
@@ -4444,10 +4343,6 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
                obd->u.obt.obt_magic = OBT_MAGIC;
        }
 
-       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.rsi_uid = 0;
        m->mdt_squash.rsi_gid = 0;
        INIT_LIST_HEAD(&m->mdt_squash.rsi_nosquash_nids);
@@ -4518,8 +4413,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",
@@ -4527,15 +4420,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)
@@ -4597,8 +4486,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. */
@@ -4616,9 +4503,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:
@@ -4739,7 +4623,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);
@@ -4790,13 +4674,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,
@@ -4837,6 +4721,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);
@@ -4925,9 +4810,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);
@@ -4965,15 +4847,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);
@@ -4989,90 +4862,6 @@ static int mdt_connect_internal(struct obd_export *exp,
        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);
-       }
-
-        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);
-}
-
 static int mdt_ctxt_add_dirty_flag(struct lu_env *env,
                                   struct mdt_thread_info *info,
                                   struct mdt_file_data *mfd)
@@ -5159,7 +4948,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 */
@@ -5188,6 +4977,7 @@ static int mdt_obd_disconnect(struct obd_export *exp)
         LASSERT(exp);
         class_export_get(exp);
 
+       nodemap_del_member(exp);
        rc = server_disconnect_export(exp);
        if (rc != 0)
                CDEBUG(D_IOCTL, "server disconnect error: rc = %d\n", rc);
@@ -5197,6 +4987,103 @@ static int mdt_obd_disconnect(struct obd_export *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);
+
+       /*
+        * 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 = 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) {
+                       rc = nodemap_add_member(*client_nid, lexp);
+                       if (rc != 0 && rc != -EEXIST)
+                               goto out;
+
+                       mdt_export_stats_init(obd, lexp, localdata);
+               }
+
+               /* For phase I, sync for cross-ref operation. */
+               spin_lock(&lexp->exp_lock);
+               lexp->exp_keep_sync = 1;
+               spin_unlock(&lexp->exp_lock);
+       }
+out:
+       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)
+{
+       lnet_nid_t             *client_nid = 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) {
+               rc = nodemap_add_member(*client_nid, exp);
+               if (rc == 0 || rc == -EEXIST)
+                       mdt_export_stats_init(obd, exp, localdata);
+       }
+
+       RETURN(rc);
+}
+
 /* FIXME: Can we avoid using these two interfaces? */
 static int mdt_init_export(struct obd_export *exp)
 {
@@ -5258,25 +5145,6 @@ static int mdt_destroy_export(struct obd_export *exp)
        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)
 {
@@ -5308,8 +5176,21 @@ 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
+ *
+ * \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 mdt_device       *mdt = info->mti_mdt;
        struct mdt_object       *mdt_obj;
@@ -5320,8 +5201,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
@@ -5331,16 +5213,14 @@ 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_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);
-       *tmpfid = pli->pli_fids[0];
+       *tmpfid = fp->gf_fid = *mdt_object_fid(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])) {
+       while (!lu_fid_eq(&mdt->mdt_md_root_fid, &fp->gf_fid)) {
                struct lu_buf           lmv_buf;
 
                mdt_obj = mdt_object_find(info->mti_env, mdt, tmpfid);
@@ -5368,18 +5248,17 @@ static int mdt_path_current(struct mdt_thread_info *info,
                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;
@@ -5393,7 +5272,7 @@ static int mdt_path_current(struct mdt_thread_info *info,
 
                        /* For slave stripes, get its master */
                        if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE) {
-                               pli->pli_fids[pli->pli_fidcount] = *tmpfid;
+                               fp->gf_fid = *tmpfid;
                                continue;
                        }
                } else if (rc < 0 && rc != -ENODATA) {
@@ -5404,72 +5283,81 @@ static int mdt_path_current(struct mdt_thread_info *info,
 
                /* Pack the name in the end of the buffer */
                ptr -= tmpname->ln_namelen;
-               if (ptr - 1 <= pli->pli_path)
+               if (ptr - 1 <= fp->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_path, ptr, fp->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 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';
+               fp->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);
 
        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 getinfo_fid2path *fp)
 {
@@ -5492,10 +5380,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))
@@ -5512,8 +5401,7 @@ static int mdt_fid2path(struct mdt_thread_info *info,
                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);
 
        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);
@@ -5688,6 +5576,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_force_abort_recovery = 1;
                target_stop_recovery_thread(obd);
                rc = 0;
                break;
@@ -5963,6 +5852,8 @@ static void __exit mdt_mod_exit(void)
 
 MODULE_AUTHOR("Sun Microsystems, 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_mod_init);
+module_exit(mdt_mod_exit);