+ mdt_lock_handle_init(lh);
+ mdt_lock_reg_init(lh, LCK_EX);
+
+ if (mdt_object_remote(pobj)) {
+ /* don't bother to check if pobj and obj are on the same MDT. */
+ rc = mdt_remote_object_lock(info, pobj, mdt_object_fid(obj),
+ &lh->mlh_rreg_lh, LCK_EX,
+ MDS_INODELOCK_LOOKUP, false);
+ } else if (mdt_object_remote(obj)) {
+ struct ldlm_res_id *res = &info->mti_res_id;
+ union ldlm_policy_data *policy = &info->mti_policy;
+ __u64 dlmflags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB |
+ LDLM_FL_COS_INCOMPAT;
+
+ fid_build_reg_res_name(mdt_object_fid(obj), res);
+ memset(policy, 0, sizeof(*policy));
+ policy->l_inodebits.bits = MDS_INODELOCK_LOOKUP;
+ rc = mdt_fid_lock(info->mti_env, info->mti_mdt->mdt_namespace,
+ &lh->mlh_reg_lh, LCK_EX, policy, res,
+ dlmflags, NULL);
+ } else {
+ /* do nothing if both are local */
+ return 0;
+ }
+
+ if (rc != ELDLM_OK)
+ return rc;
+
+ /*
+ * TODO, currently we don't save this lock because there is no place to
+ * hold this lock handle, but to avoid race we need to save this lock.
+ */
+ mdt_object_unlock(info, NULL, lh, 1);
+
+ return 0;
+}
+
+/*
+ * operation may takes locks of linkea, or directory stripes, group them in
+ * different list.
+ */
+struct mdt_sub_lock {
+ struct mdt_object *msl_obj;
+ struct mdt_lock_handle msl_lh;
+ struct list_head msl_linkage;
+};
+
+static void mdt_unlock_list(struct mdt_thread_info *info,
+ struct list_head *list, int decref)
+{
+ struct mdt_sub_lock *msl;
+ struct mdt_sub_lock *tmp;
+
+ list_for_each_entry_safe(msl, tmp, list, msl_linkage) {
+ mdt_object_unlock_put(info, msl->msl_obj, &msl->msl_lh, decref);
+ list_del(&msl->msl_linkage);
+ OBD_FREE_PTR(msl);
+ }
+}
+
+static inline void mdt_migrate_object_unlock(struct mdt_thread_info *info,
+ struct mdt_object *obj,
+ struct mdt_lock_handle *lh,
+ struct ldlm_enqueue_info *einfo,
+ struct list_head *slave_locks,
+ int decref)
+{
+ if (mdt_object_remote(obj)) {
+ mdt_unlock_list(info, slave_locks, decref);
+ mdt_object_unlock(info, obj, lh, decref);
+ } else {
+ mdt_reint_striped_unlock(info, obj, lh, einfo, decref);
+ }
+}
+
+/*
+ * lock parents of links, and also check whether total locks don't exceed
+ * RS_MAX_LOCKS.
+ *
+ * \retval 0 on success, and locks can be saved in ptlrpc_reply_stat
+ * \retval 1 on success, but total lock count may exceed RS_MAX_LOCKS
+ * \retval -ev negative errno upon error
+ */
+static int mdt_link_parents_lock(struct mdt_thread_info *info,
+ struct mdt_object *pobj,
+ const struct md_attr *ma,
+ struct mdt_object *obj,
+ struct mdt_lock_handle *lhp,
+ struct ldlm_enqueue_info *peinfo,
+ struct list_head *parent_slave_locks,
+ struct list_head *link_locks)
+{
+ struct mdt_device *mdt = info->mti_mdt;
+ struct lu_buf *buf = &info->mti_big_buf;
+ struct lu_name *lname = &info->mti_name;
+ struct linkea_data ldata = { NULL };
+ bool blocked = false;
+ int local_lnkp_cnt = 0;
+ int rc;
+
+ ENTRY;
+
+ if (S_ISDIR(lu_object_attr(&obj->mot_obj)))