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