lustre/mdt/mdt_reint.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2014, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mdt/mdt_reint.c
  37  *
  38  * Lustre Metadata Target (mdt) reintegration routines
  39  *
  40  * Author: Peter Braam <braam@clusterfs.com>
  41  * Author: Andreas Dilger <adilger@clusterfs.com>
  42  * Author: Phil Schwan <phil@clusterfs.com>
  43  * Author: Huang Hua <huanghua@clusterfs.com>
  44  * Author: Yury Umanets <umka@clusterfs.com>
  45  */
  46
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include <lprocfs_status.h>
  50 #include "mdt_internal.h"
  51 #include <lustre_lmv.h>
  52
  53 static inline void mdt_reint_init_ma(struct mdt_thread_info *info,
  54                                      struct md_attr *ma)
  55 {
  56         ma->ma_need = MA_INODE;
  57         ma->ma_valid = 0;
  58 }
  59
  60 /**
  61  * Get version of object by fid.
  62  *
  63  * Return real version or ENOENT_VERSION if object doesn't exist
  64  */
  65 static void mdt_obj_version_get(struct mdt_thread_info *info,
  66                                 struct mdt_object *o, __u64 *version)
  67 {
  68         LASSERT(o);
  69         if (mdt_object_exists(o) && !mdt_object_remote(o) &&
  70             !fid_is_obf(mdt_object_fid(o)))
  71                 *version = dt_version_get(info->mti_env, mdt_obj2dt(o));
  72         else
  73                 *version = ENOENT_VERSION;
  74         CDEBUG(D_INODE, "FID "DFID" version is "LPX64"\n",
  75                PFID(mdt_object_fid(o)), *version);
  76 }
  77
  78 /**
  79  * Check version is correct.
  80  *
  81  * Should be called only during replay.
  82  */
  83 static int mdt_version_check(struct ptlrpc_request *req,
  84                              __u64 version, int idx)
  85 {
  86         __u64 *pre_ver = lustre_msg_get_versions(req->rq_reqmsg);
  87         ENTRY;
  88
  89         if (!exp_connect_vbr(req->rq_export))
  90                 RETURN(0);
  91
  92         LASSERT(req_is_replay(req));
  93         /** VBR: version is checked always because costs nothing */
  94         LASSERT(idx < PTLRPC_NUM_VERSIONS);
  95         /** Sanity check for malformed buffers */
  96         if (pre_ver == NULL) {
  97                 CERROR("No versions in request buffer\n");
  98                 spin_lock(&req->rq_export->exp_lock);
  99                 req->rq_export->exp_vbr_failed = 1;
 100                 spin_unlock(&req->rq_export->exp_lock);
 101                 RETURN(-EOVERFLOW);
 102         } else if (pre_ver[idx] != version) {
 103                 CDEBUG(D_INODE, "Version mismatch "LPX64" != "LPX64"\n",
 104                        pre_ver[idx], version);
 105                 spin_lock(&req->rq_export->exp_lock);
 106                 req->rq_export->exp_vbr_failed = 1;
 107                 spin_unlock(&req->rq_export->exp_lock);
 108                 RETURN(-EOVERFLOW);
 109         }
 110         RETURN(0);
 111 }
 112
 113 /**
 114  * Save pre-versions in reply.
 115  */
 116 static void mdt_version_save(struct ptlrpc_request *req, __u64 version,
 117                              int idx)
 118 {
 119         __u64 *reply_ver;
 120
 121         if (!exp_connect_vbr(req->rq_export))
 122                 return;
 123
 124         LASSERT(!req_is_replay(req));
 125         LASSERT(req->rq_repmsg != NULL);
 126         reply_ver = lustre_msg_get_versions(req->rq_repmsg);
 127         if (reply_ver)
 128                 reply_ver[idx] = version;
 129 }
 130
 131 /**
 132  * Save enoent version, it is needed when it is obvious that object doesn't
 133  * exist, e.g. child during create.
 134  */
 135 static void mdt_enoent_version_save(struct mdt_thread_info *info, int idx)
 136 {
 137         /* save version of file name for replay, it must be ENOENT here */
 138         if (!req_is_replay(mdt_info_req(info))) {
 139                 info->mti_ver[idx] = ENOENT_VERSION;
 140                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 141         }
 142 }
 143
 144 /**
 145  * Get version from disk and save in reply buffer.
 146  *
 147  * Versions are saved in reply only during normal operations not replays.
 148  */
 149 void mdt_version_get_save(struct mdt_thread_info *info,
 150                           struct mdt_object *mto, int idx)
 151 {
 152         /* don't save versions during replay */
 153         if (!req_is_replay(mdt_info_req(info))) {
 154                 mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 155                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 156         }
 157 }
 158
 159 /**
 160  * Get version from disk and check it, no save in reply.
 161  */
 162 int mdt_version_get_check(struct mdt_thread_info *info,
 163                           struct mdt_object *mto, int idx)
 164 {
 165         /* only check versions during replay */
 166         if (!req_is_replay(mdt_info_req(info)))
 167                 return 0;
 168
 169         mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 170         return mdt_version_check(mdt_info_req(info), info->mti_ver[idx], idx);
 171 }
 172
 173 /**
 174  * Get version from disk and check if recovery or just save.
 175  */
 176 int mdt_version_get_check_save(struct mdt_thread_info *info,
 177                                struct mdt_object *mto, int idx)
 178 {
 179         int rc = 0;
 180
 181         mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 182         if (req_is_replay(mdt_info_req(info)))
 183                 rc = mdt_version_check(mdt_info_req(info), info->mti_ver[idx],
 184                                        idx);
 185         else
 186                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 187         return rc;
 188 }
 189
 190 /**
 191  * Lookup with version checking.
 192  *
 193  * This checks version of 'name'. Many reint functions uses 'name' for child not
 194  * FID, therefore we need to get object by name and check its version.
 195  */
 196 static int mdt_lookup_version_check(struct mdt_thread_info *info,
 197                                     struct mdt_object *p,
 198                                     const struct lu_name *lname,
 199                                     struct lu_fid *fid, int idx)
 200 {
 201         int rc, vbrc;
 202
 203         rc = mdo_lookup(info->mti_env, mdt_object_child(p), lname, fid,
 204                         &info->mti_spec);
 205         /* Check version only during replay */
 206         if (!req_is_replay(mdt_info_req(info)))
 207                 return rc;
 208
 209         info->mti_ver[idx] = ENOENT_VERSION;
 210         if (rc == 0) {
 211                 struct mdt_object *child;
 212                 child = mdt_object_find(info->mti_env, info->mti_mdt, fid);
 213                 if (likely(!IS_ERR(child))) {
 214                         mdt_obj_version_get(info, child, &info->mti_ver[idx]);
 215                         mdt_object_put(info->mti_env, child);
 216                 }
 217         }
 218         vbrc = mdt_version_check(mdt_info_req(info), info->mti_ver[idx], idx);
 219         return vbrc ? vbrc : rc;
 220
 221 }
 222
 223 /**
 224  * mdt_remote_permission: Check whether the remote operation is permitted,
 225  *
 226  * Before we implement async cross-MDT updates (DNE phase 2). There are a few
 227  * limitations here:
 228  *
 229  * 1.Only sysadmin can create remote directory and striped directory and
 230  *   migrate directory now, unless
 231  *   lctl set_param mdt.*.enable_remote_dir_gid=allow_gid.
 232  * 2.Remote directory can only be created on MDT0, unless
 233  *   lctl set_param mdt.*.enable_remote_dir = 1
 234  * 3.Only new clients can access remote dir( >= 2.4) and striped dir(>= 2.6),
 235  *   old client will return -ENOTSUPP.
 236  *
 237  * XXX these check are only needed for remote synchronization, once async
 238  * update is supported, these check will be removed.
 239  *
 240  * param[in]info:       execution environment.
 241  * param[in]parent:     the directory of this operation.
 242  * param[in]child:      the child of this operation.
 243  *
 244  * retval       = 0 remote operation is allowed.
 245  *              < 0 remote operation is denied.
 246  */
 247 static int mdt_remote_permission(struct mdt_thread_info *info,
 248                                  struct mdt_object *parent,
 249                                  struct mdt_object *child)
 250 {
 251         struct mdt_device       *mdt = info->mti_mdt;
 252         struct lu_ucred         *uc  = mdt_ucred(info);
 253         struct md_op_spec       *spec = &info->mti_spec;
 254         struct lu_attr          *attr = &info->mti_attr.ma_attr;
 255         struct obd_export       *exp = mdt_info_req(info)->rq_export;
 256
 257         /* Only check create remote directory, striped directory and
 258          * migration */
 259         if (mdt_object_remote(parent) == 0 && mdt_object_remote(child) == 0 &&
 260             !(S_ISDIR(attr->la_mode) && spec->u.sp_ea.eadata != NULL &&
 261                                         spec->u.sp_ea.eadatalen != 0) &&
 262             info->mti_rr.rr_opcode != REINT_MIGRATE)
 263                 return 0;
 264
 265         if (!md_capable(uc, CFS_CAP_SYS_ADMIN)) {
 266                 if (uc->uc_gid != mdt->mdt_enable_remote_dir_gid &&
 267                     mdt->mdt_enable_remote_dir_gid != -1)
 268                         return -EPERM;
 269         }
 270
 271         if (!mdt_is_dne_client(exp))
 272                 return -ENOTSUPP;
 273
 274         if (S_ISDIR(attr->la_mode) && spec->u.sp_ea.eadata != NULL &&
 275             spec->u.sp_ea.eadatalen != 0) {
 276                 const struct lmv_user_md *lum = spec->u.sp_ea.eadata;
 277
 278                 if (le32_to_cpu(lum->lum_stripe_count) > 1 &&
 279                     !mdt_is_striped_client(exp))
 280                         return -ENOTSUPP;
 281         }
 282
 283         return 0;
 284 }
 285
 286 /*
 287  * VBR: we save three versions in reply:
 288  * 0 - parent. Check that parent version is the same during replay.
 289  * 1 - name. Version of 'name' if file exists with the same name or
 290  * ENOENT_VERSION, it is needed because file may appear due to missed replays.
 291  * 2 - child. Version of child by FID. Must be ENOENT. It is mostly sanity
 292  * check.
 293  */
 294 static int mdt_md_create(struct mdt_thread_info *info)
 295 {
 296         struct mdt_device       *mdt = info->mti_mdt;
 297         struct mdt_object       *parent;
 298         struct mdt_object       *child;
 299         struct mdt_lock_handle  *lh;
 300         struct mdt_body         *repbody;
 301         struct md_attr          *ma = &info->mti_attr;
 302         struct mdt_reint_record *rr = &info->mti_rr;
 303         int rc;
 304         ENTRY;
 305
 306         DEBUG_REQ(D_INODE, mdt_info_req(info), "Create  ("DNAME"->"DFID") "
 307                   "in "DFID,
 308                   PNAME(&rr->rr_name), PFID(rr->rr_fid2), PFID(rr->rr_fid1));
 309
 310         if (!fid_is_md_operative(rr->rr_fid1))
 311                 RETURN(-EPERM);
 312
 313         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 314
 315         parent = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 316         if (IS_ERR(parent))
 317                 RETURN(PTR_ERR(parent));
 318
 319         if (!mdt_object_exists(parent))
 320                 GOTO(put_parent, rc = -ENOENT);
 321
 322         lh = &info->mti_lh[MDT_LH_PARENT];
 323         mdt_lock_pdo_init(lh, LCK_PW, &rr->rr_name);
 324         rc = mdt_object_lock(info, parent, lh, MDS_INODELOCK_UPDATE);
 325         if (rc)
 326                 GOTO(put_parent, rc);
 327
 328         if (!mdt_object_remote(parent)) {
 329                 rc = mdt_version_get_check_save(info, parent, 0);
 330                 if (rc)
 331                         GOTO(unlock_parent, rc);
 332         }
 333
 334         /*
 335          * Check child name version during replay.
 336          * During create replay a file may exist with same name.
 337          */
 338         rc = mdt_lookup_version_check(info, parent, &rr->rr_name,
 339                                       &info->mti_tmp_fid1, 1);
 340         if (rc == 0)
 341                 GOTO(unlock_parent, rc = -EEXIST);
 342
 343         /* -ENOENT is expected here */
 344         if (rc != -ENOENT)
 345                 GOTO(unlock_parent, rc);
 346
 347         /* save version of file name for replay, it must be ENOENT here */
 348         mdt_enoent_version_save(info, 1);
 349
 350         child = mdt_object_new(info->mti_env, mdt, rr->rr_fid2);
 351         if (likely(!IS_ERR(child))) {
 352                 struct md_object *next = mdt_object_child(parent);
 353
 354                 rc = mdt_remote_permission(info, parent, child);
 355                 if (rc != 0)
 356                         GOTO(out_put_child, rc);
 357
 358                 ma->ma_need = MA_INODE;
 359                 ma->ma_valid = 0;
 360
 361                 mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 362                                OBD_FAIL_MDS_REINT_CREATE_WRITE);
 363
 364                 /* Version of child will be updated on disk. */
 365                 tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(child));
 366                 rc = mdt_version_get_check_save(info, child, 2);
 367                 if (rc)
 368                         GOTO(out_put_child, rc);
 369
 370                 /* Let lower layer know current lock mode. */
 371                 info->mti_spec.sp_cr_mode =
 372                         mdt_dlm_mode2mdl_mode(lh->mlh_pdo_mode);
 373
 374                 /*
 375                  * Do not perform lookup sanity check. We know that name does
 376                  * not exist.
 377                  */
 378                 info->mti_spec.sp_cr_lookup = 0;
 379                 info->mti_spec.sp_feat = &dt_directory_features;
 380
 381                 rc = mdo_create(info->mti_env, next, &rr->rr_name,
 382                                 mdt_object_child(child), &info->mti_spec, ma);
 383                 if (rc == 0)
 384                         rc = mdt_attr_get_complex(info, child, ma);
 385
 386                 if (rc == 0) {
 387                         /* Return fid & attr to client. */
 388                         if (ma->ma_valid & MA_INODE)
 389                                 mdt_pack_attr2body(info, repbody, &ma->ma_attr,
 390                                                    mdt_object_fid(child));
 391                 }
 392 out_put_child:
 393                 mdt_object_put(info->mti_env, child);
 394         } else {
 395                 rc = PTR_ERR(child);
 396         }
 397 unlock_parent:
 398         mdt_object_unlock(info, parent, lh, rc);
 399 put_parent:
 400         mdt_object_put(info->mti_env, parent);
 401         RETURN(rc);
 402 }
 403
 404 static int mdt_unlock_slaves(struct mdt_thread_info *mti,
 405                              struct mdt_object *obj, __u64 ibits,
 406                              struct mdt_lock_handle *s0_lh,
 407                              struct mdt_object *s0_obj,
 408                              struct ldlm_enqueue_info *einfo)
 409 {
 410         ldlm_policy_data_t      *policy = &mti->mti_policy;
 411         int                     rc;
 412         ENTRY;
 413
 414         if (!S_ISDIR(obj->mot_header.loh_attr))
 415                 RETURN(0);
 416
 417         /* Unlock stripe 0 */
 418         if (s0_lh != NULL && lustre_handle_is_used(&s0_lh->mlh_reg_lh)) {
 419                 LASSERT(s0_obj != NULL);
 420                 mdt_object_unlock_put(mti, s0_obj, s0_lh, 1);
 421         }
 422
 423         memset(policy, 0, sizeof(*policy));
 424         policy->l_inodebits.bits = ibits;
 425
 426         rc = mo_object_unlock(mti->mti_env, mdt_object_child(obj), einfo,
 427                               policy);
 428         RETURN(rc);
 429 }
 430
 431 /**
 432  * Lock slave stripes if necessary, the lock handles of slave stripes
 433  * will be stored in einfo->ei_cbdata.
 434  **/
 435 static int mdt_lock_slaves(struct mdt_thread_info *mti, struct mdt_object *obj,
 436                            ldlm_mode_t mode, __u64 ibits,
 437                            struct mdt_lock_handle *s0_lh,
 438                            struct mdt_object **s0_objp,
 439                            struct ldlm_enqueue_info *einfo)
 440 {
 441         ldlm_policy_data_t      *policy = &mti->mti_policy;
 442         struct lu_buf           *buf = &mti->mti_buf;
 443         struct lmv_mds_md_v1    *lmv;
 444         struct lu_fid           *fid = &mti->mti_tmp_fid1;
 445         int rc;
 446         ENTRY;
 447
 448         if (!S_ISDIR(obj->mot_header.loh_attr))
 449                 RETURN(0);
 450
 451         buf->lb_buf = mti->mti_xattr_buf;
 452         buf->lb_len = sizeof(mti->mti_xattr_buf);
 453         rc = mo_xattr_get(mti->mti_env, mdt_object_child(obj), buf,
 454                           XATTR_NAME_LMV);
 455         if (rc == -ERANGE) {
 456                 rc = mdt_big_xattr_get(mti, obj, XATTR_NAME_LMV);
 457                 if (rc > 0) {
 458                         buf->lb_buf = mti->mti_big_lmm;
 459                         buf->lb_len = mti->mti_big_lmmsize;
 460                 }
 461         }
 462
 463         if (rc == -ENODATA || rc == -ENOENT)
 464                 RETURN(0);
 465
 466         if (rc <= 0)
 467                 RETURN(rc);
 468
 469         lmv = buf->lb_buf;
 470         if (le32_to_cpu(lmv->lmv_magic) != LMV_MAGIC_V1)
 471                 RETURN(-EINVAL);
 472
 473         /* Sigh, 0_stripe and master object are different
 474          * object, though they are in the same MDT, to avoid
 475          * adding osd_object_lock here, so we will enqueue the
 476          * stripe0 lock in MDT0 for now */
 477         fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[0]);
 478         *s0_objp = mdt_object_find_lock(mti, fid, s0_lh, ibits);
 479         if (IS_ERR(*s0_objp))
 480                 RETURN(PTR_ERR(*s0_objp));
 481
 482         memset(einfo, 0, sizeof(*einfo));
 483         einfo->ei_type = LDLM_IBITS;
 484         einfo->ei_mode = mode;
 485         einfo->ei_cb_bl = mdt_remote_blocking_ast;
 486         einfo->ei_cb_local_bl = mdt_blocking_ast;
 487         einfo->ei_cb_cp = ldlm_completion_ast;
 488         einfo->ei_enq_slave = 1;
 489         einfo->ei_namespace = mti->mti_mdt->mdt_namespace;
 490         memset(policy, 0, sizeof(*policy));
 491         policy->l_inodebits.bits = ibits;
 492
 493         rc = mo_object_lock(mti->mti_env, mdt_object_child(obj), NULL, einfo,
 494                             policy);
 495         RETURN(rc);
 496 }
 497
 498 static int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
 499                         struct md_attr *ma)
 500 {
 501         struct mdt_lock_handle  *lh;
 502         int do_vbr = ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID|LA_FLAGS);
 503         __u64 lockpart = MDS_INODELOCK_UPDATE;
 504         struct ldlm_enqueue_info *einfo = &info->mti_einfo;
 505         struct mdt_lock_handle  *s0_lh;
 506         struct mdt_object       *s0_obj = NULL;
 507         int rc;
 508         ENTRY;
 509
 510         lh = &info->mti_lh[MDT_LH_PARENT];
 511         mdt_lock_reg_init(lh, LCK_PW);
 512
 513         /* Even though the new MDT will grant PERM lock to the old
 514          * client, but the old client will almost ignore that during
 515          * So it needs to revoke both LOOKUP and PERM lock here, so
 516          * both new and old client can cancel the dcache */
 517         if (ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID))
 518                 lockpart |= MDS_INODELOCK_LOOKUP | MDS_INODELOCK_PERM;
 519
 520         rc = mdt_object_lock(info, mo, lh, lockpart);
 521         if (rc != 0)
 522                 RETURN(rc);
 523
 524         s0_lh = &info->mti_lh[MDT_LH_LOCAL];
 525         mdt_lock_reg_init(s0_lh, LCK_PW);
 526         rc = mdt_lock_slaves(info, mo, LCK_PW, lockpart, s0_lh, &s0_obj, einfo);
 527         if (rc != 0)
 528                 GOTO(out_unlock, rc);
 529
 530         /* all attrs are packed into mti_attr in unpack_setattr */
 531         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 532                        OBD_FAIL_MDS_REINT_SETATTR_WRITE);
 533
 534         /* This is only for set ctime when rename's source is on remote MDS. */
 535         if (unlikely(ma->ma_attr.la_valid == LA_CTIME))
 536                 ma->ma_attr_flags |= MDS_VTX_BYPASS;
 537
 538         /* VBR: update version if attr changed are important for recovery */
 539         if (do_vbr) {
 540                 /* update on-disk version of changed object */
 541                 tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(mo));
 542                 rc = mdt_version_get_check_save(info, mo, 0);
 543                 if (rc)
 544                         GOTO(out_unlock, rc);
 545         }
 546
 547         /* Ensure constant striping during chown(). See LU-2789. */
 548         if (ma->ma_attr.la_valid & (LA_UID|LA_GID))
 549                 mutex_lock(&mo->mot_lov_mutex);
 550
 551         /* all attrs are packed into mti_attr in unpack_setattr */
 552         rc = mo_attr_set(info->mti_env, mdt_object_child(mo), ma);
 553
 554         if (ma->ma_attr.la_valid & (LA_UID|LA_GID))
 555                 mutex_unlock(&mo->mot_lov_mutex);
 556
 557         if (rc != 0)
 558                 GOTO(out_unlock, rc);
 559
 560         EXIT;
 561 out_unlock:
 562         mdt_unlock_slaves(info, mo, lockpart, s0_lh, s0_obj, einfo);
 563         mdt_object_unlock(info, mo, lh, rc);
 564         return rc;
 565 }
 566
 567 /**
 568  * Check HSM flags and add HS_DIRTY flag if relevant.
 569  *
 570  * A file could be set dirty only if it has a copy in the backend (HS_EXISTS)
 571  * and is not RELEASED.
 572  */
 573 int mdt_add_dirty_flag(struct mdt_thread_info *info, struct mdt_object *mo,
 574                         struct md_attr *ma)
 575 {
 576         int rc;
 577         ENTRY;
 578
 579         /* If the file was modified, add the dirty flag */
 580         ma->ma_need = MA_HSM;
 581         rc = mdt_attr_get_complex(info, mo, ma);
 582         if (rc) {
 583                 CERROR("file attribute read error for "DFID": %d.\n",
 584                         PFID(mdt_object_fid(mo)), rc);
 585                 RETURN(rc);
 586         }
 587
 588         /* If an up2date copy exists in the backend, add dirty flag */
 589         if ((ma->ma_valid & MA_HSM) && (ma->ma_hsm.mh_flags & HS_EXISTS)
 590             && !(ma->ma_hsm.mh_flags & (HS_DIRTY|HS_RELEASED))) {
 591                 struct mdt_lock_handle  *lh = &info->mti_lh[MDT_LH_CHILD];
 592
 593                 ma->ma_hsm.mh_flags |= HS_DIRTY;
 594
 595                 mdt_lock_reg_init(lh, LCK_PW);
 596                 rc = mdt_object_lock(info, mo, lh, MDS_INODELOCK_XATTR);
 597                 if (rc != 0)
 598                         RETURN(rc);
 599
 600                 rc = mdt_hsm_attr_set(info, mo, &ma->ma_hsm);
 601                 if (rc)
 602                         CERROR("file attribute change error for "DFID": %d\n",
 603                                 PFID(mdt_object_fid(mo)), rc);
 604                 mdt_object_unlock(info, mo, lh, rc);
 605         }
 606
 607         RETURN(rc);
 608 }
 609
 610 static int mdt_reint_setattr(struct mdt_thread_info *info,
 611                              struct mdt_lock_handle *lhc)
 612 {
 613         struct md_attr          *ma = &info->mti_attr;
 614         struct mdt_reint_record *rr = &info->mti_rr;
 615         struct ptlrpc_request   *req = mdt_info_req(info);
 616         struct mdt_object       *mo;
 617         struct mdt_body         *repbody;
 618         int                      rc, rc2;
 619         ENTRY;
 620
 621         DEBUG_REQ(D_INODE, req, "setattr "DFID" %x", PFID(rr->rr_fid1),
 622                   (unsigned int)ma->ma_attr.la_valid);
 623
 624         if (info->mti_dlm_req)
 625                 ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
 626
 627         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 628         mo = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 629         if (IS_ERR(mo))
 630                 GOTO(out, rc = PTR_ERR(mo));
 631
 632         if (!mdt_object_exists(mo))
 633                 GOTO(out_put, rc = -ENOENT);
 634
 635         if (mdt_object_remote(mo))
 636                 GOTO(out_put, rc = -EREMOTE);
 637
 638         if ((ma->ma_attr.la_valid & LA_SIZE) ||
 639             (rr->rr_flags & MRF_OPEN_TRUNC)) {
 640                 /* Check write access for the O_TRUNC case */
 641                 if (mdt_write_read(mo) < 0)
 642                         GOTO(out_put, rc = -ETXTBSY);
 643         }
 644
 645         if ((ma->ma_valid & MA_INODE) && ma->ma_attr.la_valid) {
 646                 if (ma->ma_valid & MA_LOV)
 647                         GOTO(out_put, rc = -EPROTO);
 648
 649                 rc = mdt_attr_set(info, mo, ma);
 650                 if (rc)
 651                         GOTO(out_put, rc);
 652         } else if ((ma->ma_valid & MA_LOV) && (ma->ma_valid & MA_INODE)) {
 653                 struct lu_buf *buf  = &info->mti_buf;
 654
 655                 if (ma->ma_attr.la_valid != 0)
 656                         GOTO(out_put, rc = -EPROTO);
 657
 658                 buf->lb_buf = ma->ma_lmm;
 659                 buf->lb_len = ma->ma_lmm_size;
 660                 rc = mo_xattr_set(info->mti_env, mdt_object_child(mo),
 661                                   buf, XATTR_NAME_LOV, 0);
 662                 if (rc)
 663                         GOTO(out_put, rc);
 664         } else if ((ma->ma_valid & MA_LMV) && (ma->ma_valid & MA_INODE)) {
 665                 struct lu_buf *buf  = &info->mti_buf;
 666                 struct mdt_lock_handle  *lh;
 667
 668                 if (ma->ma_attr.la_valid != 0)
 669                         GOTO(out_put, rc = -EPROTO);
 670
 671                 lh = &info->mti_lh[MDT_LH_PARENT];
 672                 mdt_lock_reg_init(lh, LCK_PW);
 673
 674                 rc = mdt_object_lock(info, mo, lh,
 675                                      MDS_INODELOCK_XATTR);
 676                 if (rc != 0)
 677                         GOTO(out_put, rc);
 678
 679                 buf->lb_buf = ma->ma_lmv;
 680                 buf->lb_len = ma->ma_lmv_size;
 681                 rc = mo_xattr_set(info->mti_env, mdt_object_child(mo),
 682                                   buf, XATTR_NAME_DEFAULT_LMV, 0);
 683
 684                 mdt_object_unlock(info, mo, lh, rc);
 685                 if (rc)
 686                         GOTO(out_put, rc);
 687         } else {
 688                 GOTO(out_put, rc = -EPROTO);
 689         }
 690
 691         /* If file data is modified, add the dirty flag */
 692         if (ma->ma_attr_flags & MDS_DATA_MODIFIED)
 693                 rc = mdt_add_dirty_flag(info, mo, ma);
 694
 695         ma->ma_need = MA_INODE;
 696         ma->ma_valid = 0;
 697         rc = mdt_attr_get_complex(info, mo, ma);
 698         if (rc != 0)
 699                 GOTO(out_put, rc);
 700
 701         mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
 702
 703         EXIT;
 704 out_put:
 705         mdt_object_put(info->mti_env, mo);
 706 out:
 707         if (rc == 0)
 708                 mdt_counter_incr(req, LPROC_MDT_SETATTR);
 709
 710         mdt_client_compatibility(info);
 711         rc2 = mdt_fix_reply(info);
 712         if (rc == 0)
 713                 rc = rc2;
 714         return rc;
 715 }
 716
 717 static int mdt_reint_create(struct mdt_thread_info *info,
 718                             struct mdt_lock_handle *lhc)
 719 {
 720         struct ptlrpc_request   *req = mdt_info_req(info);
 721         int                     rc;
 722         ENTRY;
 723
 724         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 725                 RETURN(err_serious(-ESTALE));
 726
 727         if (info->mti_dlm_req)
 728                 ldlm_request_cancel(mdt_info_req(info),
 729                                     info->mti_dlm_req, 0, LATF_SKIP);
 730
 731         if (!lu_name_is_valid(&info->mti_rr.rr_name))
 732                 RETURN(-EPROTO);
 733
 734         switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
 735         case S_IFDIR:
 736                 mdt_counter_incr(req, LPROC_MDT_MKDIR);
 737                 break;
 738         case S_IFREG:
 739         case S_IFLNK:
 740         case S_IFCHR:
 741         case S_IFBLK:
 742         case S_IFIFO:
 743         case S_IFSOCK:
 744                 /* Special file should stay on the same node as parent. */
 745                 mdt_counter_incr(req, LPROC_MDT_MKNOD);
 746                 break;
 747         default:
 748                 CERROR("%s: Unsupported mode %o\n",
 749                        mdt_obd_name(info->mti_mdt),
 750                        info->mti_attr.ma_attr.la_mode);
 751                 RETURN(err_serious(-EOPNOTSUPP));
 752         }
 753
 754         rc = mdt_md_create(info);
 755         RETURN(rc);
 756 }
 757
 758 /*
 759  * VBR: save parent version in reply and child version getting by its name.
 760  * Version of child is getting and checking during its lookup. If
 761  */
 762 static int mdt_reint_unlink(struct mdt_thread_info *info,
 763                             struct mdt_lock_handle *lhc)
 764 {
 765         struct mdt_reint_record *rr = &info->mti_rr;
 766         struct ptlrpc_request   *req = mdt_info_req(info);
 767         struct md_attr          *ma = &info->mti_attr;
 768         struct lu_fid           *child_fid = &info->mti_tmp_fid1;
 769         struct mdt_object       *mp;
 770         struct mdt_object       *mc;
 771         struct mdt_lock_handle  *parent_lh;
 772         struct mdt_lock_handle  *child_lh;
 773         struct ldlm_enqueue_info *einfo = &info->mti_einfo;
 774         struct mdt_lock_handle  *s0_lh = NULL;
 775         struct mdt_object       *s0_obj = NULL;
 776         __u64                   lock_ibits;
 777         int                     rc;
 778         int                     no_name = 0;
 779         ENTRY;
 780
 781         DEBUG_REQ(D_INODE, req, "unlink "DFID"/"DNAME"", PFID(rr->rr_fid1),
 782                   PNAME(&rr->rr_name));
 783
 784         if (info->mti_dlm_req)
 785                 ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
 786
 787         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 788                 RETURN(err_serious(-ENOENT));
 789
 790         if (!fid_is_md_operative(rr->rr_fid1))
 791                 RETURN(-EPERM);
 792
 793         /*
 794          * step 1: Found the parent.
 795          */
 796         mp = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 797         if (IS_ERR(mp)) {
 798                 rc = PTR_ERR(mp);
 799                 GOTO(out, rc);
 800         }
 801
 802         parent_lh = &info->mti_lh[MDT_LH_PARENT];
 803         mdt_lock_pdo_init(parent_lh, LCK_PW, &rr->rr_name);
 804         rc = mdt_object_lock(info, mp, parent_lh, MDS_INODELOCK_UPDATE);
 805         if (rc != 0)
 806                 GOTO(put_parent, rc);
 807
 808         if (!mdt_object_remote(mp)) {
 809                 rc = mdt_version_get_check_save(info, mp, 0);
 810                 if (rc)
 811                         GOTO(unlock_parent, rc);
 812         }
 813
 814         /* step 2: find & lock the child */
 815         /* lookup child object along with version checking */
 816         fid_zero(child_fid);
 817         rc = mdt_lookup_version_check(info, mp, &rr->rr_name, child_fid, 1);
 818         if (rc != 0) {
 819                 /* Name might not be able to find during resend of
 820                  * remote unlink, considering following case.
 821                  * dir_A is a remote directory, the name entry of
 822                  * dir_A is on MDT0, the directory is on MDT1,
 823                  *
 824                  * 1. client sends unlink req to MDT1.
 825                  * 2. MDT1 sends name delete update to MDT0.
 826                  * 3. name entry is being deleted in MDT0 synchronously.
 827                  * 4. MDT1 is restarted.
 828                  * 5. client resends unlink req to MDT1. So it can not
 829                  *    find the name entry on MDT0 anymore.
 830                  * In this case, MDT1 only needs to destory the local
 831                  * directory.
 832                  * */
 833                 if (mdt_object_remote(mp) && rc == -ENOENT &&
 834                     !fid_is_zero(rr->rr_fid2) &&
 835                     lustre_msg_get_flags(req->rq_reqmsg) & MSG_RESENT) {
 836                         no_name = 1;
 837                         *child_fid = *rr->rr_fid2;
 838                  } else {
 839                         GOTO(unlock_parent, rc);
 840                  }
 841         }
 842
 843         if (!fid_is_md_operative(child_fid))
 844                 GOTO(unlock_parent, rc = -EPERM);
 845
 846         /* We will lock the child regardless it is local or remote. No harm. */
 847         mc = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
 848         if (IS_ERR(mc))
 849                 GOTO(unlock_parent, rc = PTR_ERR(mc));
 850
 851         child_lh = &info->mti_lh[MDT_LH_CHILD];
 852         mdt_lock_reg_init(child_lh, LCK_EX);
 853         if (mdt_object_remote(mc)) {
 854                 struct mdt_body  *repbody;
 855
 856                 if (!fid_is_zero(rr->rr_fid2)) {
 857                         CDEBUG(D_INFO, "%s: name "DNAME" cannot find "DFID"\n",
 858                                mdt_obd_name(info->mti_mdt),
 859                                PNAME(&rr->rr_name), PFID(mdt_object_fid(mc)));
 860                         GOTO(put_child, rc = -ENOENT);
 861                 }
 862                 CDEBUG(D_INFO, "%s: name "DNAME": "DFID" is on another MDT\n",
 863                        mdt_obd_name(info->mti_mdt),
 864                        PNAME(&rr->rr_name), PFID(mdt_object_fid(mc)));
 865
 866                 if (!mdt_is_dne_client(req->rq_export))
 867                         /* Return -ENOTSUPP for old client */
 868                         GOTO(put_child, rc = -ENOTSUPP);
 869
 870                 if (info->mti_spec.sp_rm_entry) {
 871                         struct lu_ucred *uc  = mdt_ucred(info);
 872
 873                         if (!md_capable(uc, CFS_CAP_SYS_ADMIN)) {
 874                                 CERROR("%s: unlink remote entry is only "
 875                                        "permitted for administrator: rc = %d\n",
 876                                         mdt_obd_name(info->mti_mdt),
 877                                         -EPERM);
 878                                 GOTO(put_child, rc = -EPERM);
 879                         }
 880
 881                         ma->ma_need = MA_INODE;
 882                         ma->ma_valid = 0;
 883                         rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
 884                                         NULL, &rr->rr_name, ma, no_name);
 885                         GOTO(put_child, rc);
 886                 }
 887                 /* Revoke the LOOKUP lock of the remote object granted by
 888                  * this MDT. Since the unlink will happen on another MDT,
 889                  * it will release the LOOKUP lock right away. Then What
 890                  * would happen if another client try to grab the LOOKUP
 891                  * lock at the same time with unlink XXX */
 892                 mdt_object_lock(info, mc, child_lh, MDS_INODELOCK_LOOKUP);
 893                 repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 894                 LASSERT(repbody != NULL);
 895                 repbody->mbo_fid1 = *mdt_object_fid(mc);
 896                 repbody->mbo_valid |= (OBD_MD_FLID | OBD_MD_MDS);
 897                 GOTO(unlock_child, rc = -EREMOTE);
 898         } else if (info->mti_spec.sp_rm_entry) {
 899                 rc = -EPERM;
 900                 CDEBUG(D_INFO, "%s: no rm_entry on local dir '"DNAME"': "
 901                        "rc = %d\n",
 902                        mdt_obd_name(info->mti_mdt), PNAME(&rr->rr_name), rc);
 903                 GOTO(put_child, rc);
 904         }
 905
 906         /* We used to acquire MDS_INODELOCK_FULL here but we can't do
 907          * this now because a running HSM restore on the child (unlink
 908          * victim) will hold the layout lock. See LU-4002. */
 909         lock_ibits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE;
 910         if (mdt_object_remote(mp)) {
 911                 /* Enqueue lookup lock from parent MDT */
 912                 rc = mdt_remote_object_lock(info, mp, mdt_object_fid(mc),
 913                                             &child_lh->mlh_rreg_lh,
 914                                             child_lh->mlh_rreg_mode,
 915                                             MDS_INODELOCK_LOOKUP);
 916                 if (rc != ELDLM_OK)
 917                         GOTO(put_child, rc);
 918
 919                 lock_ibits &= ~MDS_INODELOCK_LOOKUP;
 920         }
 921
 922         rc = mdt_object_lock(info, mc, child_lh, lock_ibits);
 923         if (rc != 0)
 924                 GOTO(put_child, rc);
 925         /*
 926          * Now we can only make sure we need MA_INODE, in mdd layer, will check
 927          * whether need MA_LOV and MA_COOKIE.
 928          */
 929         ma->ma_need = MA_INODE;
 930         ma->ma_valid = 0;
 931
 932         s0_lh = &info->mti_lh[MDT_LH_LOCAL];
 933         mdt_lock_reg_init(s0_lh, LCK_EX);
 934         rc = mdt_lock_slaves(info, mc, LCK_EX, MDS_INODELOCK_UPDATE, s0_lh,
 935                              &s0_obj, einfo);
 936         if (rc != 0)
 937                 GOTO(unlock_child, rc);
 938
 939         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 940                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
 941         /* save version when object is locked */
 942         mdt_version_get_save(info, mc, 1);
 943
 944         mutex_lock(&mc->mot_lov_mutex);
 945
 946         rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
 947                         mdt_object_child(mc), &rr->rr_name, ma, no_name);
 948
 949         mutex_unlock(&mc->mot_lov_mutex);
 950
 951         if (rc == 0 && !lu_object_is_dying(&mc->mot_header))
 952                 rc = mdt_attr_get_complex(info, mc, ma);
 953         if (rc == 0)
 954                 mdt_handle_last_unlink(info, mc, ma);
 955
 956         if (ma->ma_valid & MA_INODE) {
 957                 switch (ma->ma_attr.la_mode & S_IFMT) {
 958                 case S_IFDIR:
 959                         mdt_counter_incr(req, LPROC_MDT_RMDIR);
 960                         break;
 961                 case S_IFREG:
 962                 case S_IFLNK:
 963                 case S_IFCHR:
 964                 case S_IFBLK:
 965                 case S_IFIFO:
 966                 case S_IFSOCK:
 967                         mdt_counter_incr(req, LPROC_MDT_UNLINK);
 968                         break;
 969                 default:
 970                         LASSERTF(0, "bad file type %o unlinking\n",
 971                                  ma->ma_attr.la_mode);
 972                 }
 973         }
 974
 975         EXIT;
 976
 977 unlock_child:
 978         mdt_unlock_slaves(info, mc, MDS_INODELOCK_UPDATE, s0_lh, s0_obj, einfo);
 979         mdt_object_unlock(info, mc, child_lh, rc);
 980 put_child:
 981         mdt_object_put(info->mti_env, mc);
 982 unlock_parent:
 983         mdt_object_unlock(info, mp, parent_lh, rc);
 984 put_parent:
 985         mdt_object_put(info->mti_env, mp);
 986 out:
 987         return rc;
 988 }
 989
 990 /*
 991  * VBR: save versions in reply: 0 - parent; 1 - child by fid; 2 - target by
 992  * name.
 993  */
 994 static int mdt_reint_link(struct mdt_thread_info *info,
 995                           struct mdt_lock_handle *lhc)
 996 {
 997         struct mdt_reint_record *rr = &info->mti_rr;
 998         struct ptlrpc_request   *req = mdt_info_req(info);
 999         struct md_attr          *ma = &info->mti_attr;
1000         struct mdt_object       *ms;
1001         struct mdt_object       *mp;
1002         struct mdt_lock_handle  *lhs;
1003         struct mdt_lock_handle  *lhp;
1004         int rc;
1005         ENTRY;
1006
1007         DEBUG_REQ(D_INODE, req, "link "DFID" to "DFID"/"DNAME,
1008                   PFID(rr->rr_fid1), PFID(rr->rr_fid2), PNAME(&rr->rr_name));
1009
1010         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
1011                 RETURN(err_serious(-ENOENT));
1012
1013         if (info->mti_dlm_req)
1014                 ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
1015
1016         /* Invalid case so return error immediately instead of
1017          * processing it */
1018         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2))
1019                 RETURN(-EPERM);
1020
1021         if (!fid_is_md_operative(rr->rr_fid1) ||
1022             !fid_is_md_operative(rr->rr_fid2))
1023                 RETURN(-EPERM);
1024
1025         /* step 1: find & lock the target parent dir */
1026         lhp = &info->mti_lh[MDT_LH_PARENT];
1027         mdt_lock_pdo_init(lhp, LCK_PW, &rr->rr_name);
1028         mp = mdt_object_find_lock(info, rr->rr_fid2, lhp,
1029                                   MDS_INODELOCK_UPDATE);
1030         if (IS_ERR(mp))
1031                 RETURN(PTR_ERR(mp));
1032
1033         rc = mdt_version_get_check_save(info, mp, 0);
1034         if (rc)
1035                 GOTO(out_unlock_parent, rc);
1036
1037         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME3, 5);
1038
1039         /* step 2: find & lock the source */
1040         lhs = &info->mti_lh[MDT_LH_CHILD];
1041         mdt_lock_reg_init(lhs, LCK_EX);
1042
1043         ms = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
1044         if (IS_ERR(ms))
1045                 GOTO(out_unlock_parent, rc = PTR_ERR(ms));
1046
1047         if (!mdt_object_exists(ms)) {
1048                 mdt_object_put(info->mti_env, ms);
1049                 CDEBUG(D_INFO, "%s: "DFID" does not exist.\n",
1050                        mdt_obd_name(info->mti_mdt), PFID(rr->rr_fid1));
1051                 GOTO(out_unlock_parent, rc = -ENOENT);
1052         }
1053
1054         rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE |
1055                              MDS_INODELOCK_XATTR);
1056         if (rc != 0) {
1057                 mdt_object_put(info->mti_env, ms);
1058                 GOTO(out_unlock_parent, rc);
1059         }
1060
1061         /* step 3: link it */
1062         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
1063                        OBD_FAIL_MDS_REINT_LINK_WRITE);
1064
1065         tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(ms));
1066         rc = mdt_version_get_check_save(info, ms, 1);
1067         if (rc)
1068                 GOTO(out_unlock_child, rc);
1069
1070         /** check target version by name during replay */
1071         rc = mdt_lookup_version_check(info, mp, &rr->rr_name,
1072                                       &info->mti_tmp_fid1, 2);
1073         if (rc != 0 && rc != -ENOENT)
1074                 GOTO(out_unlock_child, rc);
1075         /* save version of file name for replay, it must be ENOENT here */
1076         if (!req_is_replay(mdt_info_req(info))) {
1077                 if (rc != -ENOENT) {
1078                         CDEBUG(D_INFO, "link target "DNAME" existed!\n",
1079                                PNAME(&rr->rr_name));
1080                         GOTO(out_unlock_child, rc = -EEXIST);
1081                 }
1082                 info->mti_ver[2] = ENOENT_VERSION;
1083                 mdt_version_save(mdt_info_req(info), info->mti_ver[2], 2);
1084         }
1085
1086         rc = mdo_link(info->mti_env, mdt_object_child(mp),
1087               mdt_object_child(ms), &rr->rr_name, ma);
1088
1089         if (rc == 0)
1090                 mdt_counter_incr(req, LPROC_MDT_LINK);
1091
1092         EXIT;
1093 out_unlock_child:
1094         mdt_object_unlock_put(info, ms, lhs, rc);
1095 out_unlock_parent:
1096         mdt_object_unlock_put(info, mp, lhp, rc);
1097         return rc;
1098 }
1099 /**
1100  * lock the part of the directory according to the hash of the name
1101  * (lh->mlh_pdo_hash) in parallel directory lock.
1102  */
1103 static int mdt_pdir_hash_lock(struct mdt_thread_info *info,
1104                               struct mdt_lock_handle *lh,
1105                               struct mdt_object *obj, __u64 ibits)
1106 {
1107         struct ldlm_res_id *res = &info->mti_res_id;
1108         struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
1109         ldlm_policy_data_t *policy = &info->mti_policy;
1110         int rc;
1111
1112         /*
1113          * Finish res_id initializing by name hash marking part of
1114          * directory which is taking modification.
1115          */
1116         LASSERT(lh->mlh_pdo_hash != 0);
1117         fid_build_pdo_res_name(mdt_object_fid(obj), lh->mlh_pdo_hash, res);
1118         memset(policy, 0, sizeof(*policy));
1119         policy->l_inodebits.bits = ibits;
1120         /*
1121          * Use LDLM_FL_LOCAL_ONLY for this lock. We do not know yet if it is
1122          * going to be sent to client. If it is - mdt_intent_policy() path will
1123          * fix it up and turn FL_LOCAL flag off.
1124          */
1125         rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy,
1126                           res, LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB,
1127                           &info->mti_exp->exp_handle.h_cookie);
1128         return rc;
1129 }
1130
1131 /**
1132  * Get BFL lock for rename or migrate process.
1133  **/
1134 static int mdt_rename_lock(struct mdt_thread_info *info,
1135                            struct lustre_handle *lh)
1136 {
1137         int     rc;
1138         ENTRY;
1139
1140         if (mdt_seq_site(info->mti_mdt)->ss_node_id != 0) {
1141                 struct lu_fid *fid = &info->mti_tmp_fid1;
1142                 struct mdt_object *obj;
1143
1144                 /* XXX, right now, it has to use object API to
1145                  * enqueue lock cross MDT, so it will enqueue
1146                  * rename lock(with LUSTRE_BFL_FID) by root object */
1147                 lu_root_fid(fid);
1148                 obj = mdt_object_find(info->mti_env, info->mti_mdt, fid);
1149                 if (IS_ERR(obj))
1150                         RETURN(PTR_ERR(obj));
1151
1152                 rc = mdt_remote_object_lock(info, obj,
1153                                             &LUSTRE_BFL_FID, lh,
1154                                             LCK_EX,
1155                                             MDS_INODELOCK_UPDATE);
1156                 mdt_object_put(info->mti_env, obj);
1157         } else {
1158                 struct ldlm_namespace   *ns = info->mti_mdt->mdt_namespace;
1159                 ldlm_policy_data_t      *policy = &info->mti_policy;
1160                 struct ldlm_res_id      *res_id = &info->mti_res_id;
1161                 __u64                   flags = 0;
1162
1163                 fid_build_reg_res_name(&LUSTRE_BFL_FID, res_id);
1164                 memset(policy, 0, sizeof *policy);
1165                 policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
1166                 flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
1167                 rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, policy,
1168                                            LCK_EX, &flags, ldlm_blocking_ast,
1169                                            ldlm_completion_ast, NULL, NULL, 0,
1170                                            LVB_T_NONE,
1171                                            &info->mti_exp->exp_handle.h_cookie,
1172                                            lh);
1173                 RETURN(rc);
1174         }
1175         RETURN(rc);
1176 }
1177
1178 static void mdt_rename_unlock(struct lustre_handle *lh)
1179 {
1180         ENTRY;
1181         LASSERT(lustre_handle_is_used(lh));
1182         /* Cancel the single rename lock right away */
1183         ldlm_lock_decref_and_cancel(lh, LCK_EX);
1184         EXIT;
1185 }
1186
1187 /*
1188  * This is is_subdir() variant, it is CMD if cmm forwards it to correct
1189  * target. Source should not be ancestor of target dir. May be other rename
1190  * checks can be moved here later.
1191  */
1192 static int mdt_is_subdir(struct mdt_thread_info *info,
1193                          struct mdt_object *dir,
1194                          const struct lu_fid *fid)
1195 {
1196         struct lu_fid dir_fid = dir->mot_header.loh_fid;
1197         int rc = 0;
1198         ENTRY;
1199
1200         /* If the source and target are in the same directory, they can not
1201          * be parent/child relationship, so subdir check is not needed */
1202         if (lu_fid_eq(&dir_fid, fid))
1203                 return 0;
1204
1205         if (!mdt_object_exists(dir))
1206                 RETURN(-ENOENT);
1207
1208         rc = mdo_is_subdir(info->mti_env, mdt_object_child(dir),
1209                            fid, &dir_fid);
1210         if (rc < 0) {
1211                 CERROR("%s: failed subdir check in "DFID" for "DFID
1212                        ": rc = %d\n", mdt_obd_name(info->mti_mdt),
1213                        PFID(&dir_fid), PFID(fid), rc);
1214                 /* Return EINVAL only if a parent is the @fid */
1215                 if (rc == -EINVAL)
1216                         rc = -EIO;
1217         } else {
1218                 /* check the found fid */
1219                 if (lu_fid_eq(&dir_fid, fid))
1220                         rc = -EINVAL;
1221         }
1222
1223         RETURN(rc);
1224 }
1225
1226 /* Update object linkEA */
1227 struct mdt_lock_list {
1228         struct mdt_object       *mll_obj;
1229         struct mdt_lock_handle  mll_lh;
1230         struct list_head        mll_list;
1231 };
1232
1233 static void mdt_unlock_list(struct mdt_thread_info *info,
1234                             struct list_head *list, int rc)
1235 {
1236         struct mdt_lock_list *mll;
1237         struct mdt_lock_list *mll2;
1238
1239         list_for_each_entry_safe(mll, mll2, list, mll_list) {
1240                 mdt_object_unlock_put(info, mll->mll_obj, &mll->mll_lh, rc);
1241                 list_del(&mll->mll_list);
1242                 OBD_FREE_PTR(mll);
1243         }
1244 }
1245
1246 static int mdt_lock_objects_in_linkea(struct mdt_thread_info *info,
1247                                       struct mdt_object *obj,
1248                                       struct mdt_object *pobj,
1249                                       struct list_head *lock_list)
1250 {
1251         struct lu_buf           *buf = &info->mti_big_buf;
1252         struct linkea_data      ldata = { NULL };
1253         int                     count;
1254         int                     rc;
1255         ENTRY;
1256
1257         if (S_ISDIR(lu_object_attr(&obj->mot_obj)))
1258                 RETURN(0);
1259
1260         buf = lu_buf_check_and_alloc(buf, PATH_MAX);
1261         if (buf->lb_buf == NULL)
1262                 RETURN(-ENOMEM);
1263
1264         ldata.ld_buf = buf;
1265         rc = mdt_links_read(info, obj, &ldata);
1266         if (rc != 0) {
1267                 if (rc == -ENOENT || rc == -ENODATA)
1268                         rc = 0;
1269                 RETURN(rc);
1270         }
1271
1272         LASSERT(ldata.ld_leh != NULL);
1273         ldata.ld_lee = (struct link_ea_entry *)(ldata.ld_leh + 1);
1274         for (count = 0; count < ldata.ld_leh->leh_reccount; count++) {
1275                 struct mdt_device *mdt = info->mti_mdt;
1276                 struct mdt_object *mdt_pobj;
1277                 struct mdt_lock_list *mll;
1278                 struct lu_name name;
1279                 struct lu_fid  fid;
1280
1281                 linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
1282                                     &name, &fid);
1283                 ldata.ld_lee = (struct link_ea_entry *)((char *)ldata.ld_lee +
1284                                                          ldata.ld_reclen);
1285                 mdt_pobj = mdt_object_find(info->mti_env, mdt, &fid);
1286                 if (IS_ERR(mdt_pobj)) {
1287                         CWARN("%s: cannot find obj "DFID": rc = %ld\n",
1288                               mdt_obd_name(mdt), PFID(&fid), PTR_ERR(mdt_pobj));
1289                         continue;
1290                 }
1291
1292                 if (!mdt_object_exists(mdt_pobj)) {
1293                         CDEBUG(D_INFO, "%s: obj "DFID" does not exist\n",
1294                               mdt_obd_name(mdt), PFID(&fid));
1295                         mdt_object_put(info->mti_env, mdt_pobj);
1296                         continue;
1297                 }
1298
1299                 if (mdt_pobj == pobj) {
1300                         CDEBUG(D_INFO, "%s: skipping parent obj "DFID"\n",
1301                                mdt_obd_name(mdt), PFID(&fid));
1302                         mdt_object_put(info->mti_env, mdt_pobj);
1303                         continue;
1304                 }
1305
1306                 OBD_ALLOC_PTR(mll);
1307                 if (mll == NULL) {
1308                         mdt_object_put(info->mti_env, mdt_pobj);
1309                         GOTO(out, rc = -ENOMEM);
1310                 }
1311
1312                 mdt_lock_pdo_init(&mll->mll_lh, LCK_PW, &name);
1313                 rc = mdt_object_lock(info, mdt_pobj, &mll->mll_lh,
1314                                      MDS_INODELOCK_UPDATE);
1315                 if (rc != 0) {
1316                         CERROR("%s: cannot lock "DFID": rc =%d\n",
1317                                mdt_obd_name(mdt), PFID(&fid), rc);
1318                         mdt_object_put(info->mti_env, mdt_pobj);
1319                         OBD_FREE_PTR(mll);
1320                         GOTO(out, rc);
1321                 }
1322
1323                 INIT_LIST_HEAD(&mll->mll_list);
1324                 mll->mll_obj = mdt_pobj;
1325                 list_add_tail(&mll->mll_list, lock_list);
1326         }
1327 out:
1328         if (rc != 0)
1329                 mdt_unlock_list(info, lock_list, rc);
1330         RETURN(rc);
1331 }
1332
1333 /* migrate files from one MDT to another MDT */
1334 static int mdt_reint_migrate_internal(struct mdt_thread_info *info,
1335                                       struct mdt_lock_handle *lhc)
1336 {
1337         struct mdt_reint_record *rr = &info->mti_rr;
1338         struct md_attr          *ma = &info->mti_attr;
1339         struct mdt_object       *msrcdir;
1340         struct mdt_object       *mold;
1341         struct mdt_object       *mnew = NULL;
1342         struct mdt_lock_handle  *lh_dirp;
1343         struct mdt_lock_handle  *lh_childp;
1344         struct mdt_lock_handle  *lh_tgtp = NULL;
1345         struct lu_fid           *old_fid = &info->mti_tmp_fid1;
1346         struct list_head        lock_list;
1347         __u64                   lock_ibits;
1348         struct ldlm_lock        *lease = NULL;
1349         bool                    lock_open_sem = false;
1350         int                     rc;
1351         ENTRY;
1352
1353         CDEBUG(D_INODE, "migrate "DFID"/"DNAME" to "DFID"\n", PFID(rr->rr_fid1),
1354                PNAME(&rr->rr_name), PFID(rr->rr_fid2));
1355
1356         /* 1: lock the source dir. */
1357         msrcdir = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
1358         if (IS_ERR(msrcdir)) {
1359                 CERROR("%s: cannot find source dir "DFID" : rc = %d\n",
1360                         mdt_obd_name(info->mti_mdt), PFID(rr->rr_fid1),
1361                         (int)PTR_ERR(msrcdir));
1362                 RETURN(PTR_ERR(msrcdir));
1363         }
1364
1365         lh_dirp = &info->mti_lh[MDT_LH_PARENT];
1366         mdt_lock_pdo_init(lh_dirp, LCK_PW, &rr->rr_name);
1367         rc = mdt_object_lock(info, msrcdir, lh_dirp,
1368                              MDS_INODELOCK_UPDATE);
1369         if (rc)
1370                 GOTO(out_put_parent, rc);
1371
1372         if (!mdt_object_remote(msrcdir)) {
1373                 rc = mdt_version_get_check_save(info, msrcdir, 0);
1374                 if (rc)
1375                         GOTO(out_unlock_parent, rc);
1376         }
1377
1378         /* 2: sanity check and find the object to be migrated. */
1379         fid_zero(old_fid);
1380         rc = mdt_lookup_version_check(info, msrcdir, &rr->rr_name, old_fid, 2);
1381         if (rc != 0)
1382                 GOTO(out_unlock_parent, rc);
1383
1384         if (lu_fid_eq(old_fid, rr->rr_fid1) || lu_fid_eq(old_fid, rr->rr_fid2))
1385                 GOTO(out_unlock_parent, rc = -EINVAL);
1386
1387         if (!fid_is_md_operative(old_fid))
1388                 GOTO(out_unlock_parent, rc = -EPERM);
1389
1390         mold = mdt_object_find(info->mti_env, info->mti_mdt, old_fid);
1391         if (IS_ERR(mold))
1392                 GOTO(out_unlock_parent, rc = PTR_ERR(mold));
1393
1394         if (mdt_object_remote(mold)) {
1395                 CERROR("%s: source "DFID" is on the remote MDT\n",
1396                        mdt_obd_name(info->mti_mdt), PFID(old_fid));
1397                 GOTO(out_put_child, rc = -EREMOTE);
1398         }
1399
1400         if (S_ISREG(lu_object_attr(&mold->mot_obj)) &&
1401             !mdt_object_remote(msrcdir)) {
1402                 CERROR("%s: parent "DFID" is still on the same"
1403                        " MDT, which should be migrated first:"
1404                        " rc = %d\n", mdt_obd_name(info->mti_mdt),
1405                        PFID(mdt_object_fid(msrcdir)), -EPERM);
1406                 GOTO(out_put_child, rc = -EPERM);
1407         }
1408
1409         rc = mdt_remote_permission(info, msrcdir, mold);
1410         if (rc != 0)
1411                 GOTO(out_put_child, rc);
1412
1413         /* 3: iterate the linkea of the object and lock all of the objects */
1414         INIT_LIST_HEAD(&lock_list);
1415         rc = mdt_lock_objects_in_linkea(info, mold, msrcdir, &lock_list);
1416         if (rc != 0)
1417                 GOTO(out_put_child, rc);
1418
1419         if (info->mti_spec.sp_migrate_close) {
1420                 struct close_data *data;
1421                 struct mdt_body  *repbody;
1422                 bool lease_broken = false;
1423
1424                 if (!req_capsule_field_present(info->mti_pill, &RMF_MDT_EPOCH,
1425                                       RCL_CLIENT) ||
1426                     !req_capsule_field_present(info->mti_pill, &RMF_CLOSE_DATA,
1427                                       RCL_CLIENT))
1428                         GOTO(out_lease, rc = -EPROTO);
1429
1430                 data = req_capsule_client_get(info->mti_pill, &RMF_CLOSE_DATA);
1431                 if (data == NULL)
1432                         GOTO(out_lease, rc = -EPROTO);
1433
1434                 lease = ldlm_handle2lock(&data->cd_handle);
1435                 if (lease == NULL)
1436                         GOTO(out_lease, rc = -ESTALE);
1437
1438                 /* try to hold open_sem so that nobody else can open the file */
1439                 if (!down_write_trylock(&mold->mot_open_sem)) {
1440                         ldlm_lock_cancel(lease);
1441                         GOTO(out_lease, rc = -EBUSY);
1442                 }
1443
1444                 lock_open_sem = true;
1445                 /* Check if the lease open lease has already canceled */
1446                 lock_res_and_lock(lease);
1447                 lease_broken = ldlm_is_cancel(lease);
1448                 unlock_res_and_lock(lease);
1449
1450                 LDLM_DEBUG(lease, DFID " lease broken? %d\n",
1451                            PFID(mdt_object_fid(mold)), lease_broken);
1452
1453                 /* Cancel server side lease. Client side counterpart should
1454                  * have been cancelled. It's okay to cancel it now as we've
1455                  * held mot_open_sem. */
1456                 ldlm_lock_cancel(lease);
1457
1458                 if (lease_broken)
1459                         GOTO(out_lease, rc = -EAGAIN);
1460 out_lease:
1461                 rc = mdt_close_internal(info, mdt_info_req(info), NULL);
1462                 repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
1463                 repbody->mbo_valid |= OBD_MD_CLOSE_INTENT_EXECED;
1464                 if (rc != 0)
1465                         GOTO(out_unlock_list, rc);
1466         }
1467
1468         /* 4: lock of the object migrated object */
1469         lh_childp = &info->mti_lh[MDT_LH_OLD];
1470         mdt_lock_reg_init(lh_childp, LCK_EX);
1471         lock_ibits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE |
1472                      MDS_INODELOCK_LAYOUT;
1473         if (mdt_object_remote(msrcdir)) {
1474                 /* Enqueue lookup lock from the parent MDT */
1475                 rc = mdt_remote_object_lock(info, msrcdir, mdt_object_fid(mold),
1476                                             &lh_childp->mlh_rreg_lh,
1477                                             lh_childp->mlh_rreg_mode,
1478                                             MDS_INODELOCK_LOOKUP);
1479                 if (rc != ELDLM_OK)
1480                         GOTO(out_unlock_list, rc);
1481
1482                 lock_ibits &= ~MDS_INODELOCK_LOOKUP;
1483         }
1484
1485         rc = mdt_object_lock(info, mold, lh_childp, lock_ibits);
1486         if (rc != 0)
1487                 GOTO(out_unlock_child, rc);
1488
1489         ma->ma_need = MA_LMV;
1490         ma->ma_valid = 0;
1491         ma->ma_lmv = (union lmv_mds_md *)info->mti_xattr_buf;
1492         ma->ma_lmv_size = sizeof(info->mti_xattr_buf);
1493         rc = mdt_stripe_get(info, mold, ma, XATTR_NAME_LMV);
1494         if (rc != 0)
1495                 GOTO(out_unlock_child, rc);
1496
1497         if ((ma->ma_valid & MA_LMV)) {
1498                 struct lmv_mds_md_v1 *lmm1;
1499
1500                 lmv_le_to_cpu(ma->ma_lmv, ma->ma_lmv);
1501                 lmm1 = &ma->ma_lmv->lmv_md_v1;
1502                 if (!(lmm1->lmv_hash_type & LMV_HASH_FLAG_MIGRATION)) {
1503                         CERROR("%s: can not migrate striped dir "DFID
1504                                ": rc = %d\n", mdt_obd_name(info->mti_mdt),
1505                                PFID(mdt_object_fid(mold)), -EPERM);
1506                         GOTO(out_unlock_child, rc = -EPERM);
1507                 }
1508
1509                 if (!fid_is_sane(&lmm1->lmv_stripe_fids[1]))
1510                         GOTO(out_unlock_child, rc = -EINVAL);
1511
1512                 mnew = mdt_object_find(info->mti_env, info->mti_mdt,
1513                                        &lmm1->lmv_stripe_fids[1]);
1514                 if (IS_ERR(mnew))
1515                         GOTO(out_unlock_child, rc = PTR_ERR(mnew));
1516
1517                 if (!mdt_object_remote(mnew)) {
1518                         CERROR("%s: "DFID" being migrated is on this MDT:"
1519                                " rc  = %d\n", mdt_obd_name(info->mti_mdt),
1520                                PFID(rr->rr_fid2), -EPERM);
1521                         GOTO(out_put_new, rc = -EPERM);
1522                 }
1523
1524                 lh_tgtp = &info->mti_lh[MDT_LH_CHILD];
1525                 mdt_lock_reg_init(lh_tgtp, LCK_EX);
1526                 rc = mdt_remote_object_lock(info, mnew,
1527                                             mdt_object_fid(mnew),
1528                                             &lh_tgtp->mlh_rreg_lh,
1529                                             lh_tgtp->mlh_rreg_mode,
1530                                             MDS_INODELOCK_UPDATE);
1531                 if (rc != 0) {
1532                         lh_tgtp = NULL;
1533                         GOTO(out_put_new, rc);
1534                 }
1535         } else {
1536                 mnew = mdt_object_find(info->mti_env, info->mti_mdt,
1537                                        rr->rr_fid2);
1538                 if (IS_ERR(mnew))
1539                         GOTO(out_unlock_child, rc = PTR_ERR(mnew));
1540                 if (!mdt_object_remote(mnew)) {
1541                         CERROR("%s: Migration "DFID" is on this MDT:"
1542                                " rc = %d\n", mdt_obd_name(info->mti_mdt),
1543                                PFID(rr->rr_fid2), -EXDEV);
1544                         GOTO(out_put_new, rc = -EXDEV);
1545                 }
1546         }
1547
1548         /* 5: migrate it */
1549         mdt_reint_init_ma(info, ma);
1550
1551         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
1552                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
1553
1554         rc = mdo_migrate(info->mti_env, mdt_object_child(msrcdir),
1555                          mdt_object_child(mold), &rr->rr_name,
1556                          mdt_object_child(mnew), ma);
1557         if (rc != 0)
1558                 GOTO(out_unlock_new, rc);
1559
1560 out_unlock_new:
1561         if (lh_tgtp != NULL)
1562                 mdt_object_unlock(info, mnew, lh_tgtp, rc);
1563 out_put_new:
1564         if (mnew)
1565                 mdt_object_put(info->mti_env, mnew);
1566 out_unlock_child:
1567         mdt_object_unlock(info, mold, lh_childp, rc);
1568 out_unlock_list:
1569         mdt_unlock_list(info, &lock_list, rc);
1570         if (lease != NULL) {
1571                 ldlm_reprocess_all(lease->l_resource);
1572                 LDLM_LOCK_PUT(lease);
1573         }
1574
1575         if (lock_open_sem)
1576                 up_write(&mold->mot_open_sem);
1577 out_put_child:
1578         mdt_object_put(info->mti_env, mold);
1579 out_unlock_parent:
1580         mdt_object_unlock(info, msrcdir, lh_dirp, rc);
1581 out_put_parent:
1582         mdt_object_put(info->mti_env, msrcdir);
1583
1584         RETURN(rc);
1585 }
1586
1587 static struct mdt_object *mdt_object_find_check(struct mdt_thread_info *info,
1588                                                 const struct lu_fid *fid,
1589                                                 int idx)
1590 {
1591         struct mdt_object *dir;
1592         int rc;
1593         ENTRY;
1594
1595         dir = mdt_object_find(info->mti_env, info->mti_mdt, fid);
1596         if (IS_ERR(dir))
1597                 RETURN(dir);
1598
1599         /* check early, the real version will be saved after locking */
1600         rc = mdt_version_get_check(info, dir, idx);
1601         if (rc)
1602                 GOTO(out_put, rc);
1603
1604         RETURN(dir);
1605 out_put:
1606         mdt_object_put(info->mti_env, dir);
1607         return ERR_PTR(rc);
1608 }
1609
1610 static int mdt_object_lock_save(struct mdt_thread_info *info,
1611                                 struct mdt_object *dir,
1612                                 struct mdt_lock_handle *lh,
1613                                 int idx)
1614 {
1615         int rc;
1616
1617         /* we lock the target dir if it is local */
1618         rc = mdt_object_lock(info, dir, lh, MDS_INODELOCK_UPDATE);
1619         if (rc != 0)
1620                 return rc;
1621
1622         /* get and save correct version after locking */
1623         mdt_version_get_save(info, dir, idx);
1624         return 0;
1625 }
1626
1627
1628 static int mdt_rename_parents_lock(struct mdt_thread_info *info,
1629                                    struct mdt_object **srcp,
1630                                    struct mdt_object **tgtp)
1631 {
1632         struct mdt_reint_record *rr = &info->mti_rr;
1633         const struct lu_fid     *fid_src = rr->rr_fid1;
1634         const struct lu_fid     *fid_tgt = rr->rr_fid2;
1635         struct mdt_lock_handle  *lh_src = &info->mti_lh[MDT_LH_PARENT];
1636         struct mdt_lock_handle  *lh_tgt = &info->mti_lh[MDT_LH_CHILD];
1637         struct mdt_object       *src;
1638         struct mdt_object       *tgt;
1639         int                      reverse = 0;
1640         int                      rc;
1641         ENTRY;
1642
1643         /* find both parents. */
1644         src = mdt_object_find_check(info, fid_src, 0);
1645         if (IS_ERR(src))
1646                 RETURN(PTR_ERR(src));
1647
1648         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME3, 5);
1649
1650         if (lu_fid_eq(fid_src, fid_tgt)) {
1651                 tgt = src;
1652                 mdt_object_get(info->mti_env, tgt);
1653         } else {
1654                 /* Check if the @src is not a child of the @tgt, otherwise a
1655                  * reverse locking must take place. */
1656                 rc = mdt_is_subdir(info, src, fid_tgt);
1657                 if (rc == -EINVAL)
1658                         reverse = 1;
1659                 else if (rc)
1660                         GOTO(err_src_put, rc);
1661
1662                 tgt = mdt_object_find_check(info, fid_tgt, 1);
1663                 if (IS_ERR(tgt))
1664                         GOTO(err_src_put, rc = PTR_ERR(tgt));
1665         }
1666
1667         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME4, 5);
1668
1669         /* lock parents in the proper order. */
1670         if (reverse) {
1671                 rc = mdt_object_lock_save(info, tgt, lh_tgt, 1);
1672                 if (rc)
1673                         GOTO(err_tgt_put, rc);
1674
1675                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME, 5);
1676
1677                 rc = mdt_object_lock_save(info, src, lh_src, 0);
1678         } else {
1679                 rc = mdt_object_lock_save(info, src, lh_src, 0);
1680                 if (rc)
1681                         GOTO(err_tgt_put, rc);
1682
1683                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME, 5);
1684
1685                 if (tgt != src)
1686                         rc = mdt_object_lock_save(info, tgt, lh_tgt, 1);
1687                 else if (lh_src->mlh_pdo_hash != lh_tgt->mlh_pdo_hash) {
1688                         rc = mdt_pdir_hash_lock(info, lh_tgt, tgt,
1689                                                 MDS_INODELOCK_UPDATE);
1690                         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK2, 10);
1691                 }
1692         }
1693         if (rc)
1694                 GOTO(err_unlock, rc);
1695
1696         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME4, 5);
1697
1698         *srcp = src;
1699         *tgtp = tgt;
1700         RETURN(0);
1701
1702 err_unlock:
1703         /* The order does not matter as the handle is checked inside,
1704          * as well as not used handle. */
1705         mdt_object_unlock(info, src, lh_src, rc);
1706         mdt_object_unlock(info, tgt, lh_tgt, rc);
1707 err_tgt_put:
1708         mdt_object_put(info->mti_env, tgt);
1709 err_src_put:
1710         mdt_object_put(info->mti_env, src);
1711         RETURN(rc);
1712 }
1713
1714 /*
1715  * VBR: rename versions in reply: 0 - src parent; 1 - tgt parent;
1716  * 2 - src child; 3 - tgt child.
1717  * Update on disk version of src child.
1718  */
1719 /**
1720  * For DNE phase I, only these renames are allowed
1721  *      mv src_p/src_c tgt_p/tgt_c
1722  * 1. src_p/src_c/tgt_p/tgt_c are in the same MDT.
1723  * 2. src_p and tgt_p are same directory, and tgt_c does not
1724  *    exists. In this case, all of modification will happen
1725  *    in the MDT where ithesource parent is, only one remote
1726  *    update is needed, i.e. set c_time/m_time on the child.
1727  *    And tgt_c will be still in the same MDT as the original
1728  *    src_c.
1729  */
1730 static int mdt_reint_rename_internal(struct mdt_thread_info *info,
1731                                      struct mdt_lock_handle *lhc)
1732 {
1733         struct mdt_reint_record *rr = &info->mti_rr;
1734         struct md_attr          *ma = &info->mti_attr;
1735         struct ptlrpc_request   *req = mdt_info_req(info);
1736         struct mdt_object       *msrcdir = NULL;
1737         struct mdt_object       *mtgtdir = NULL;
1738         struct mdt_object       *mold;
1739         struct mdt_object       *mnew = NULL;
1740         struct mdt_lock_handle  *lh_srcdirp;
1741         struct mdt_lock_handle  *lh_tgtdirp;
1742         struct mdt_lock_handle  *lh_oldp = NULL;
1743         struct mdt_lock_handle  *lh_newp = NULL;
1744         struct lu_fid           *old_fid = &info->mti_tmp_fid1;
1745         struct lu_fid           *new_fid = &info->mti_tmp_fid2;
1746         __u64                   lock_ibits;
1747         int                      rc;
1748         ENTRY;
1749
1750         DEBUG_REQ(D_INODE, req, "rename "DFID"/"DNAME" to "DFID"/"DNAME,
1751                   PFID(rr->rr_fid1), PNAME(&rr->rr_name),
1752                   PFID(rr->rr_fid2), PNAME(&rr->rr_tgt_name));
1753
1754         lh_srcdirp = &info->mti_lh[MDT_LH_PARENT];
1755         mdt_lock_pdo_init(lh_srcdirp, LCK_PW, &rr->rr_name);
1756         lh_tgtdirp = &info->mti_lh[MDT_LH_CHILD];
1757         mdt_lock_pdo_init(lh_tgtdirp, LCK_PW, &rr->rr_tgt_name);
1758
1759         /* step 1&2: lock the source and target dirs. */
1760         rc = mdt_rename_parents_lock(info, &msrcdir, &mtgtdir);
1761         if (rc)
1762                 RETURN(rc);
1763
1764         OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_RENAME2, 5);
1765
1766         /* step 3: find & lock the old object. */
1767         fid_zero(old_fid);
1768         rc = mdt_lookup_version_check(info, msrcdir, &rr->rr_name, old_fid, 2);
1769         if (rc != 0)
1770                 GOTO(out_unlock_parents, rc);
1771
1772         if (lu_fid_eq(old_fid, rr->rr_fid1) || lu_fid_eq(old_fid, rr->rr_fid2))
1773                 GOTO(out_unlock_parents, rc = -EINVAL);
1774
1775         if (!fid_is_md_operative(old_fid))
1776                 GOTO(out_unlock_parents, rc = -EPERM);
1777
1778         mold = mdt_object_find(info->mti_env, info->mti_mdt, old_fid);
1779         if (IS_ERR(mold))
1780                 GOTO(out_unlock_parents, rc = PTR_ERR(mold));
1781
1782         /* Check if @mtgtdir is subdir of @mold, before locking child
1783          * to avoid reverse locking. */
1784         rc = mdt_is_subdir(info, mtgtdir, old_fid);
1785         if (rc)
1786                 GOTO(out_put_old, rc);
1787
1788         tgt_vbr_obj_set(info->mti_env, mdt_obj2dt(mold));
1789         /* save version after locking */
1790         mdt_version_get_save(info, mold, 2);
1791
1792         /* step 4: find & lock the new object. */
1793         /* new target object may not exist now */
1794         /* lookup with version checking */
1795         fid_zero(new_fid);
1796         rc = mdt_lookup_version_check(info, mtgtdir, &rr->rr_tgt_name, new_fid,
1797                                       3);
1798         if (rc == 0) {
1799                 /* the new_fid should have been filled at this moment */
1800                 if (lu_fid_eq(old_fid, new_fid))
1801                         GOTO(out_put_old, rc);
1802
1803                 if (lu_fid_eq(new_fid, rr->rr_fid1) ||
1804                     lu_fid_eq(new_fid, rr->rr_fid2))
1805                         GOTO(out_put_old, rc = -EINVAL);
1806
1807                 if (!fid_is_md_operative(new_fid))
1808                         GOTO(out_put_old, rc = -EPERM);
1809
1810                 mnew = mdt_object_find(info->mti_env, info->mti_mdt, new_fid);
1811                 if (IS_ERR(mnew))
1812                         GOTO(out_put_old, rc = PTR_ERR(mnew));
1813
1814                 if (mdt_object_remote(mnew)) {
1815                         struct mdt_body  *repbody;
1816
1817                         /* Always send rename req to the target child MDT */
1818                         repbody = req_capsule_server_get(info->mti_pill,
1819                                                          &RMF_MDT_BODY);
1820                         LASSERT(repbody != NULL);
1821                         repbody->mbo_fid1 = *new_fid;
1822                         repbody->mbo_valid |= (OBD_MD_FLID | OBD_MD_MDS);
1823                         GOTO(out_put_old, rc = -EXDEV);
1824                 }
1825                 /* Before locking the target dir, check we do not replace
1826                  * a dir with a non-dir, otherwise it may deadlock with
1827                  * link op which tries to create a link in this dir
1828                  * back to this non-dir. */
1829                 if (S_ISDIR(lu_object_attr(&mnew->mot_obj)) &&
1830                     !S_ISDIR(lu_object_attr(&mold->mot_obj)))
1831                         GOTO(out_put_new, rc = -EISDIR);
1832
1833                 lh_oldp = &info->mti_lh[MDT_LH_OLD];
1834                 mdt_lock_reg_init(lh_oldp, LCK_EX);
1835
1836                 lock_ibits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_XATTR;
1837                 if (mdt_object_remote(msrcdir)) {
1838                         /* Enqueue lookup lock from the parent MDT */
1839                         rc = mdt_remote_object_lock(info, msrcdir,
1840                                                     mdt_object_fid(mold),
1841                                                     &lh_oldp->mlh_rreg_lh,
1842                                                     lh_oldp->mlh_rreg_mode,
1843                                                     MDS_INODELOCK_LOOKUP);
1844                         if (rc != ELDLM_OK)
1845                                 GOTO(out_put_new, rc);
1846
1847                         lock_ibits &= ~MDS_INODELOCK_LOOKUP;
1848                 }
1849
1850                 rc = mdt_object_lock(info, mold, lh_oldp, lock_ibits);
1851                 if (rc != 0)
1852                         GOTO(out_unlock_old, rc);
1853
1854                 /* Check if @msrcdir is subdir of @mnew, before locking child
1855                  * to avoid reverse locking. */
1856                 rc = mdt_is_subdir(info, msrcdir, new_fid);
1857                 if (rc)
1858                         GOTO(out_unlock_old, rc);
1859
1860                 /* We used to acquire MDS_INODELOCK_FULL here but we
1861                  * can't do this now because a running HSM restore on
1862                  * the rename onto victim will hold the layout
1863                  * lock. See LU-4002. */
1864
1865                 lh_newp = &info->mti_lh[MDT_LH_NEW];
1866                 mdt_lock_reg_init(lh_newp, LCK_EX);
1867                 rc = mdt_object_lock(info, mnew, lh_newp,
1868                                      MDS_INODELOCK_LOOKUP |
1869                                      MDS_INODELOCK_UPDATE);
1870                 if (rc != 0)
1871                         GOTO(out_unlock_old, rc);
1872
1873                 /* get and save version after locking */
1874                 mdt_version_get_save(info, mnew, 3);
1875         } else if (rc != -EREMOTE && rc != -ENOENT) {
1876                 GOTO(out_put_old, rc);
1877         } else {
1878                 lh_oldp = &info->mti_lh[MDT_LH_OLD];
1879                 mdt_lock_reg_init(lh_oldp, LCK_EX);
1880
1881                 lock_ibits = MDS_INODELOCK_LOOKUP | MDS_INODELOCK_XATTR;
1882                 if (mdt_object_remote(msrcdir)) {
1883                         /* Enqueue lookup lock from the parent MDT */
1884                         rc = mdt_remote_object_lock(info, msrcdir,
1885                                                     mdt_object_fid(mold),
1886                                                     &lh_oldp->mlh_rreg_lh,
1887                                                     lh_oldp->mlh_rreg_mode,
1888                                                     MDS_INODELOCK_LOOKUP);
1889                         if (rc != ELDLM_OK)
1890                                 GOTO(out_put_new, rc);
1891
1892                         lock_ibits &= ~MDS_INODELOCK_LOOKUP;
1893                 }
1894
1895                 rc = mdt_object_lock(info, mold, lh_oldp, lock_ibits);
1896                 if (rc != 0)
1897                         GOTO(out_put_old, rc);
1898
1899                 mdt_enoent_version_save(info, 3);
1900         }
1901
1902         /* step 5: rename it */
1903         mdt_reint_init_ma(info, ma);
1904
1905         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
1906                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
1907
1908         if (mnew != NULL)
1909                 mutex_lock(&mnew->mot_lov_mutex);
1910
1911         rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
1912                         mdt_object_child(mtgtdir), old_fid, &rr->rr_name,
1913                         mnew != NULL ? mdt_object_child(mnew) : NULL,
1914                         &rr->rr_tgt_name, ma);
1915
1916         if (mnew != NULL)
1917                 mutex_unlock(&mnew->mot_lov_mutex);
1918
1919         /* handle last link of tgt object */
1920         if (rc == 0) {
1921                 mdt_counter_incr(req, LPROC_MDT_RENAME);
1922                 if (mnew)
1923                         mdt_handle_last_unlink(info, mnew, ma);
1924
1925                 mdt_rename_counter_tally(info, info->mti_mdt, req,
1926                                          msrcdir, mtgtdir);
1927         }
1928
1929         EXIT;
1930         if (mnew != NULL)
1931                 mdt_object_unlock(info, mnew, lh_newp, rc);
1932 out_unlock_old:
1933         mdt_object_unlock(info, mold, lh_oldp, rc);
1934 out_put_new:
1935         if (mnew != NULL)
1936                 mdt_object_put(info->mti_env, mnew);
1937 out_put_old:
1938         mdt_object_put(info->mti_env, mold);
1939 out_unlock_parents:
1940         mdt_object_unlock_put(info, mtgtdir, lh_tgtdirp, rc);
1941         mdt_object_unlock_put(info, msrcdir, lh_srcdirp, rc);
1942         return rc;
1943 }
1944
1945 static int mdt_reint_rename_or_migrate(struct mdt_thread_info *info,
1946                                        struct mdt_lock_handle *lhc, bool rename)
1947 {
1948         struct mdt_reint_record *rr = &info->mti_rr;
1949         struct ptlrpc_request   *req = mdt_info_req(info);
1950         struct lustre_handle    rename_lh = { 0 };
1951         int                     rc;
1952         ENTRY;
1953
1954         if (info->mti_dlm_req)
1955                 ldlm_request_cancel(req, info->mti_dlm_req, 0, LATF_SKIP);
1956
1957         if (!fid_is_md_operative(rr->rr_fid1) ||
1958             !fid_is_md_operative(rr->rr_fid2))
1959                 RETURN(-EPERM);
1960
1961         /* Note: do not enqueue rename lock for replay request, because
1962          * if other MDT holds rename lock, but being blocked to wait for
1963          * this MDT to finish its recovery, and the failover MDT can not
1964          * get rename lock, which will cause deadlock. */
1965         if (!req_is_replay(req)) {
1966                 rc = mdt_rename_lock(info, &rename_lh);
1967                 if (rc != 0) {
1968                         CERROR("%s: can't lock FS for rename: rc  = %d\n",
1969                                mdt_obd_name(info->mti_mdt), rc);
1970                         RETURN(rc);
1971                 }
1972         }
1973
1974         if (rename)
1975                 rc = mdt_reint_rename_internal(info, lhc);
1976         else
1977                 rc = mdt_reint_migrate_internal(info, lhc);
1978
1979         if (lustre_handle_is_used(&rename_lh))
1980                 mdt_rename_unlock(&rename_lh);
1981
1982         RETURN(rc);
1983 }
1984
1985 static int mdt_reint_rename(struct mdt_thread_info *info,
1986                             struct mdt_lock_handle *lhc)
1987 {
1988         return mdt_reint_rename_or_migrate(info, lhc, true);
1989 }
1990
1991 static int mdt_reint_migrate(struct mdt_thread_info *info,
1992                             struct mdt_lock_handle *lhc)
1993 {
1994         return mdt_reint_rename_or_migrate(info, lhc, false);
1995 }
1996
1997 struct mdt_reinter {
1998         int (*mr_handler)(struct mdt_thread_info *, struct mdt_lock_handle *);
1999         enum lprocfs_extra_opc mr_extra_opc;
2000 };
2001
2002 static const struct mdt_reinter mdt_reinters[] = {
2003         [REINT_SETATTR] = {
2004                 .mr_handler = &mdt_reint_setattr,
2005                 .mr_extra_opc = MDS_REINT_SETATTR,
2006         },
2007         [REINT_CREATE] = {
2008                 .mr_handler = &mdt_reint_create,
2009                 .mr_extra_opc = MDS_REINT_CREATE,
2010         },
2011         [REINT_LINK] = {
2012                 .mr_handler = &mdt_reint_link,
2013                 .mr_extra_opc = MDS_REINT_LINK,
2014         },
2015         [REINT_UNLINK] = {
2016                 .mr_handler = &mdt_reint_unlink,
2017                 .mr_extra_opc = MDS_REINT_UNLINK,
2018         },
2019         [REINT_RENAME] = {
2020                 .mr_handler = &mdt_reint_rename,
2021                 .mr_extra_opc = MDS_REINT_RENAME,
2022         },
2023         [REINT_OPEN] = {
2024                 .mr_handler = &mdt_reint_open,
2025                 .mr_extra_opc = MDS_REINT_OPEN,
2026         },
2027         [REINT_SETXATTR] = {
2028                 .mr_handler = &mdt_reint_setxattr,
2029                 .mr_extra_opc = MDS_REINT_SETXATTR,
2030         },
2031         [REINT_RMENTRY] = {
2032                 .mr_handler = &mdt_reint_unlink,
2033                 .mr_extra_opc = MDS_REINT_UNLINK,
2034         },
2035         [REINT_MIGRATE] = {
2036                 .mr_handler = &mdt_reint_migrate,
2037                 .mr_extra_opc = MDS_REINT_RENAME,
2038         },
2039 };
2040
2041 int mdt_reint_rec(struct mdt_thread_info *info,
2042                   struct mdt_lock_handle *lhc)
2043 {
2044         const struct mdt_reinter *mr;
2045         int rc;
2046         ENTRY;
2047
2048         if (!(info->mti_rr.rr_opcode < ARRAY_SIZE(mdt_reinters)))
2049                 RETURN(-EPROTO);
2050
2051         mr = &mdt_reinters[info->mti_rr.rr_opcode];
2052         if (mr->mr_handler == NULL)
2053                 RETURN(-EPROTO);
2054
2055         rc = (*mr->mr_handler)(info, lhc);
2056
2057         lprocfs_counter_incr(ptlrpc_req2svc(mdt_info_req(info))->srv_stats,
2058                              PTLRPC_LAST_CNTR + mr->mr_extra_opc);
2059
2060         RETURN(rc);
2061 }