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, 2012, Whamcloud, Inc.
  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 "mdt_internal.h"
  50 #include "lu_time.h"
  51
  52 static inline void mdt_reint_init_ma(struct mdt_thread_info *info,
  53                                      struct md_attr *ma)
  54 {
  55         ma->ma_need = MA_INODE;
  56         ma->ma_valid = 0;
  57 }
  58
  59 static int mdt_create_pack_capa(struct mdt_thread_info *info, int rc,
  60                                 struct mdt_object *object,
  61                                 struct mdt_body *repbody)
  62 {
  63         ENTRY;
  64
  65         /* for cross-ref mkdir, mds capa has been fetched from remote obj, then
  66          * we won't go to below*/
  67         if (repbody->valid & OBD_MD_FLMDSCAPA)
  68                 RETURN(rc);
  69
  70         if (rc == 0 && info->mti_mdt->mdt_opts.mo_mds_capa &&
  71             info->mti_exp->exp_connect_flags & OBD_CONNECT_MDS_CAPA) {
  72                 struct lustre_capa *capa;
  73
  74                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA1);
  75                 LASSERT(capa);
  76                 capa->lc_opc = CAPA_OPC_MDS_DEFAULT;
  77                 rc = mo_capa_get(info->mti_env, mdt_object_child(object), capa,
  78                                  0);
  79                 if (rc == 0)
  80                         repbody->valid |= OBD_MD_FLMDSCAPA;
  81         }
  82
  83         RETURN(rc);
  84 }
  85
  86 /**
  87  * Get version of object by fid.
  88  *
  89  * Return real version or ENOENT_VERSION if object doesn't exist
  90  */
  91 static void mdt_obj_version_get(struct mdt_thread_info *info,
  92                                 struct mdt_object *o, __u64 *version)
  93 {
  94         LASSERT(o);
  95         LASSERT(mdt_object_exists(o) >= 0);
  96         if (mdt_object_exists(o) > 0 && !mdt_object_obf(o))
  97                 *version = dt_version_get(info->mti_env, mdt_obj2dt(o));
  98         else
  99                 *version = ENOENT_VERSION;
 100         CDEBUG(D_INODE, "FID "DFID" version is "LPX64"\n",
 101                PFID(mdt_object_fid(o)), *version);
 102 }
 103
 104 /**
 105  * Check version is correct.
 106  *
 107  * Should be called only during replay.
 108  */
 109 static int mdt_version_check(struct ptlrpc_request *req,
 110                              __u64 version, int idx)
 111 {
 112         __u64 *pre_ver = lustre_msg_get_versions(req->rq_reqmsg);
 113         ENTRY;
 114
 115         if (!exp_connect_vbr(req->rq_export))
 116                 RETURN(0);
 117
 118         LASSERT(req_is_replay(req));
 119         /** VBR: version is checked always because costs nothing */
 120         LASSERT(idx < PTLRPC_NUM_VERSIONS);
 121         /** Sanity check for malformed buffers */
 122         if (pre_ver == NULL) {
 123                 CERROR("No versions in request buffer\n");
 124                 spin_lock(&req->rq_export->exp_lock);
 125                 req->rq_export->exp_vbr_failed = 1;
 126                 spin_unlock(&req->rq_export->exp_lock);
 127                 RETURN(-EOVERFLOW);
 128         } else if (pre_ver[idx] != version) {
 129                 CDEBUG(D_INODE, "Version mismatch "LPX64" != "LPX64"\n",
 130                        pre_ver[idx], version);
 131                 spin_lock(&req->rq_export->exp_lock);
 132                 req->rq_export->exp_vbr_failed = 1;
 133                 spin_unlock(&req->rq_export->exp_lock);
 134                 RETURN(-EOVERFLOW);
 135         }
 136         RETURN(0);
 137 }
 138
 139 /**
 140  * Save pre-versions in reply.
 141  */
 142 static void mdt_version_save(struct ptlrpc_request *req, __u64 version,
 143                              int idx)
 144 {
 145         __u64 *reply_ver;
 146
 147         if (!exp_connect_vbr(req->rq_export))
 148                 return;
 149
 150         LASSERT(!req_is_replay(req));
 151         LASSERT(req->rq_repmsg != NULL);
 152         reply_ver = lustre_msg_get_versions(req->rq_repmsg);
 153         if (reply_ver)
 154                 reply_ver[idx] = version;
 155 }
 156
 157 /**
 158  * Save enoent version, it is needed when it is obvious that object doesn't
 159  * exist, e.g. child during create.
 160  */
 161 static void mdt_enoent_version_save(struct mdt_thread_info *info, int idx)
 162 {
 163         /* save version of file name for replay, it must be ENOENT here */
 164         if (!req_is_replay(mdt_info_req(info))) {
 165                 info->mti_ver[idx] = ENOENT_VERSION;
 166                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 167         }
 168 }
 169
 170 /**
 171  * Get version from disk and save in reply buffer.
 172  *
 173  * Versions are saved in reply only during normal operations not replays.
 174  */
 175 void mdt_version_get_save(struct mdt_thread_info *info,
 176                           struct mdt_object *mto, int idx)
 177 {
 178         /* don't save versions during replay */
 179         if (!req_is_replay(mdt_info_req(info))) {
 180                 mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 181                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 182         }
 183 }
 184
 185 /**
 186  * Get version from disk and check it, no save in reply.
 187  */
 188 int mdt_version_get_check(struct mdt_thread_info *info,
 189                           struct mdt_object *mto, int idx)
 190 {
 191         /* only check versions during replay */
 192         if (!req_is_replay(mdt_info_req(info)))
 193                 return 0;
 194
 195         mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 196         return mdt_version_check(mdt_info_req(info), info->mti_ver[idx], idx);
 197 }
 198
 199 /**
 200  * Get version from disk and check if recovery or just save.
 201  */
 202 int mdt_version_get_check_save(struct mdt_thread_info *info,
 203                                struct mdt_object *mto, int idx)
 204 {
 205         int rc = 0;
 206
 207         mdt_obj_version_get(info, mto, &info->mti_ver[idx]);
 208         if (req_is_replay(mdt_info_req(info)))
 209                 rc = mdt_version_check(mdt_info_req(info), info->mti_ver[idx],
 210                                        idx);
 211         else
 212                 mdt_version_save(mdt_info_req(info), info->mti_ver[idx], idx);
 213         return rc;
 214 }
 215
 216 /**
 217  * Lookup with version checking.
 218  *
 219  * This checks version of 'name'. Many reint functions uses 'name' for child not
 220  * FID, therefore we need to get object by name and check its version.
 221  */
 222 int mdt_lookup_version_check(struct mdt_thread_info *info,
 223                              struct mdt_object *p, struct lu_name *lname,
 224                              struct lu_fid *fid, int idx)
 225 {
 226         int rc, vbrc;
 227
 228         rc = mdo_lookup(info->mti_env, mdt_object_child(p), lname, fid,
 229                         &info->mti_spec);
 230         /* Check version only during replay */
 231         if (!req_is_replay(mdt_info_req(info)))
 232                 return rc;
 233
 234         info->mti_ver[idx] = ENOENT_VERSION;
 235         if (rc == 0) {
 236                 struct mdt_object *child;
 237                 child = mdt_object_find(info->mti_env, info->mti_mdt, fid);
 238                 if (likely(!IS_ERR(child))) {
 239                         mdt_obj_version_get(info, child, &info->mti_ver[idx]);
 240                         mdt_object_put(info->mti_env, child);
 241                 }
 242         }
 243         vbrc = mdt_version_check(mdt_info_req(info), info->mti_ver[idx], idx);
 244         return vbrc ? vbrc : rc;
 245
 246 }
 247
 248 /*
 249  * VBR: we save three versions in reply:
 250  * 0 - parent. Check that parent version is the same during replay.
 251  * 1 - name. Version of 'name' if file exists with the same name or
 252  * ENOENT_VERSION, it is needed because file may appear due to missed replays.
 253  * 2 - child. Version of child by FID. Must be ENOENT. It is mostly sanity
 254  * check.
 255  */
 256 static int mdt_md_create(struct mdt_thread_info *info)
 257 {
 258         struct mdt_device       *mdt = info->mti_mdt;
 259         struct mdt_object       *parent;
 260         struct mdt_object       *child;
 261         struct mdt_lock_handle  *lh;
 262         struct mdt_body         *repbody;
 263         struct md_attr          *ma = &info->mti_attr;
 264         struct mdt_reint_record *rr = &info->mti_rr;
 265         struct lu_name          *lname;
 266         int rc;
 267         ENTRY;
 268
 269         DEBUG_REQ(D_INODE, mdt_info_req(info), "Create  (%s->"DFID") in "DFID,
 270                   rr->rr_name, PFID(rr->rr_fid2), PFID(rr->rr_fid1));
 271
 272         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 273
 274         lh = &info->mti_lh[MDT_LH_PARENT];
 275         mdt_lock_pdo_init(lh, LCK_PW, rr->rr_name, rr->rr_namelen);
 276
 277         parent = mdt_object_find_lock(info, rr->rr_fid1, lh,
 278                                       MDS_INODELOCK_UPDATE);
 279         if (IS_ERR(parent))
 280                 RETURN(PTR_ERR(parent));
 281
 282         if (mdt_object_obf(parent))
 283                 GOTO(out_put_parent, rc = -EPERM);
 284
 285         rc = mdt_version_get_check_save(info, parent, 0);
 286         if (rc)
 287                 GOTO(out_put_parent, rc);
 288
 289         /*
 290          * Check child name version during replay.
 291          * During create replay a file may exist with same name.
 292          */
 293         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 294         rc = mdt_lookup_version_check(info, parent, lname,
 295                                       &info->mti_tmp_fid1, 1);
 296         if (rc == 0)
 297                 GOTO(out_put_parent, rc = -EEXIST);
 298
 299         /* -ENOENT is expected here */
 300         if (rc != -ENOENT)
 301                 GOTO(out_put_parent, rc);
 302
 303         /* save version of file name for replay, it must be ENOENT here */
 304         mdt_enoent_version_save(info, 1);
 305
 306         child = mdt_object_new(info->mti_env, mdt, rr->rr_fid2);
 307         if (likely(!IS_ERR(child))) {
 308                 struct md_object *next = mdt_object_child(parent);
 309
 310                 ma->ma_need = MA_INODE;
 311                 ma->ma_valid = 0;
 312                 /* capa for cross-ref will be stored here */
 313                 ma->ma_capa = req_capsule_server_get(info->mti_pill,
 314                                                      &RMF_CAPA1);
 315                 LASSERT(ma->ma_capa);
 316
 317                 mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 318                                OBD_FAIL_MDS_REINT_CREATE_WRITE);
 319
 320                 /* Version of child will be updated on disk. */
 321                 info->mti_mos = child;
 322                 rc = mdt_version_get_check_save(info, child, 2);
 323                 if (rc)
 324                         GOTO(out_put_child, rc);
 325
 326                 /* Let lower layer know current lock mode. */
 327                 info->mti_spec.sp_cr_mode =
 328                         mdt_dlm_mode2mdl_mode(lh->mlh_pdo_mode);
 329
 330                 /*
 331                  * Do not perform lookup sanity check. We know that name does
 332                  * not exist.
 333                  */
 334                 info->mti_spec.sp_cr_lookup = 0;
 335                 info->mti_spec.sp_feat = &dt_directory_features;
 336
 337                 rc = mdo_create(info->mti_env, next, lname,
 338                                 mdt_object_child(child),
 339                                 &info->mti_spec, ma);
 340                 if (rc == 0)
 341                         rc = mdt_attr_get_complex(info, child, ma);
 342
 343                 if (rc == 0) {
 344                         /* Return fid & attr to client. */
 345                         if (ma->ma_valid & MA_INODE)
 346                                 mdt_pack_attr2body(info, repbody, &ma->ma_attr,
 347                                                    mdt_object_fid(child));
 348                 }
 349 out_put_child:
 350                 mdt_object_put(info->mti_env, child);
 351         } else {
 352                 rc = PTR_ERR(child);
 353         }
 354         mdt_create_pack_capa(info, rc, child, repbody);
 355 out_put_parent:
 356         mdt_object_unlock_put(info, parent, lh, rc);
 357         RETURN(rc);
 358 }
 359
 360 int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
 361                  struct md_attr *ma, int flags)
 362 {
 363         struct mdt_lock_handle  *lh;
 364         int do_vbr = ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID|LA_FLAGS);
 365         __u64 lockpart = MDS_INODELOCK_UPDATE;
 366         int rc;
 367         ENTRY;
 368
 369         /* attr shouldn't be set on remote object */
 370         LASSERT(mdt_object_exists(mo) >= 0);
 371
 372         lh = &info->mti_lh[MDT_LH_PARENT];
 373         mdt_lock_reg_init(lh, LCK_PW);
 374
 375         if (ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID))
 376                 lockpart |= MDS_INODELOCK_LOOKUP;
 377
 378         rc = mdt_object_lock(info, mo, lh, lockpart, MDT_LOCAL_LOCK);
 379         if (rc != 0)
 380                 RETURN(rc);
 381
 382         if (mdt_object_exists(mo) == 0)
 383                 GOTO(out_unlock, rc = -ENOENT);
 384
 385         /* all attrs are packed into mti_attr in unpack_setattr */
 386         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 387                        OBD_FAIL_MDS_REINT_SETATTR_WRITE);
 388
 389         /* This is only for set ctime when rename's source is on remote MDS. */
 390         if (unlikely(ma->ma_attr.la_valid == LA_CTIME))
 391                 ma->ma_attr_flags |= MDS_VTX_BYPASS;
 392
 393         /* VBR: update version if attr changed are important for recovery */
 394         if (do_vbr) {
 395                 /* update on-disk version of changed object */
 396                 info->mti_mos = mo;
 397                 rc = mdt_version_get_check_save(info, mo, 0);
 398                 if (rc)
 399                         GOTO(out_unlock, rc);
 400         }
 401
 402         /* all attrs are packed into mti_attr in unpack_setattr */
 403         rc = mo_attr_set(info->mti_env, mdt_object_child(mo), ma);
 404         if (rc != 0)
 405                 GOTO(out_unlock, rc);
 406
 407         EXIT;
 408 out_unlock:
 409         mdt_object_unlock(info, mo, lh, rc);
 410         return rc;
 411 }
 412
 413 static int mdt_reint_setattr(struct mdt_thread_info *info,
 414                              struct mdt_lock_handle *lhc)
 415 {
 416         struct md_attr          *ma = &info->mti_attr;
 417         struct mdt_reint_record *rr = &info->mti_rr;
 418         struct ptlrpc_request   *req = mdt_info_req(info);
 419         struct mdt_export_data  *med = &req->rq_export->exp_mdt_data;
 420         struct mdt_file_data    *mfd;
 421         struct mdt_object       *mo;
 422         struct mdt_body         *repbody;
 423         int                      som_au, rc, rc2;
 424         ENTRY;
 425
 426         DEBUG_REQ(D_INODE, req, "setattr "DFID" %x", PFID(rr->rr_fid1),
 427                   (unsigned int)ma->ma_attr.la_valid);
 428
 429         if (info->mti_dlm_req)
 430                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 431
 432         repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);
 433         mo = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 434         if (IS_ERR(mo))
 435                 GOTO(out, rc = PTR_ERR(mo));
 436
 437         if (mdt_object_obf(mo))
 438                 GOTO(out_put, rc = -EPERM);
 439
 440         /* start a log jounal handle if needed */
 441         if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM)) {
 442                 if ((ma->ma_attr.la_valid & LA_SIZE) ||
 443                     (rr->rr_flags & MRF_OPEN_TRUNC)) {
 444                         /* Check write access for the O_TRUNC case */
 445                         if (mdt_write_read(mo) < 0)
 446                                 GOTO(out_put, rc = -ETXTBSY);
 447                 }
 448         } else if (info->mti_ioepoch &&
 449                    (info->mti_ioepoch->flags & MF_EPOCH_OPEN)) {
 450                 /* Truncate case. IOEpoch is opened. */
 451                 rc = mdt_write_get(mo);
 452                 if (rc)
 453                         GOTO(out_put, rc);
 454
 455                 mfd = mdt_mfd_new();
 456                 if (mfd == NULL) {
 457                         mdt_write_put(mo);
 458                         GOTO(out_put, rc = -ENOMEM);
 459                 }
 460
 461                 mdt_ioepoch_open(info, mo, 0);
 462                 repbody->ioepoch = mo->mot_ioepoch;
 463
 464                 mdt_object_get(info->mti_env, mo);
 465                 mdt_mfd_set_mode(mfd, MDS_FMODE_TRUNC);
 466                 mfd->mfd_object = mo;
 467                 mfd->mfd_xid = req->rq_xid;
 468
 469                 spin_lock(&med->med_open_lock);
 470                 cfs_list_add(&mfd->mfd_list, &med->med_open_head);
 471                 spin_unlock(&med->med_open_lock);
 472                 repbody->handle.cookie = mfd->mfd_handle.h_cookie;
 473         }
 474
 475         som_au = info->mti_ioepoch && info->mti_ioepoch->flags & MF_SOM_CHANGE;
 476         if (som_au) {
 477                 /* SOM Attribute update case. Find the proper mfd and update
 478                  * SOM attributes on the proper object. */
 479                 LASSERT(mdt_conn_flags(info) & OBD_CONNECT_SOM);
 480                 LASSERT(info->mti_ioepoch);
 481
 482                 spin_lock(&med->med_open_lock);
 483                 mfd = mdt_handle2mfd(info, &info->mti_ioepoch->handle);
 484                 if (mfd == NULL) {
 485                         spin_unlock(&med->med_open_lock);
 486                         CDEBUG(D_INODE, "no handle for file close: "
 487                                "fid = "DFID": cookie = "LPX64"\n",
 488                                PFID(info->mti_rr.rr_fid1),
 489                                info->mti_ioepoch->handle.cookie);
 490                         GOTO(out_put, rc = -ESTALE);
 491                 }
 492                 LASSERT(mfd->mfd_mode == MDS_FMODE_SOM);
 493                 LASSERT(!(info->mti_ioepoch->flags & MF_EPOCH_CLOSE));
 494
 495                 class_handle_unhash(&mfd->mfd_handle);
 496                 cfs_list_del_init(&mfd->mfd_list);
 497                 spin_unlock(&med->med_open_lock);
 498
 499                 mdt_mfd_close(info, mfd);
 500         } else if ((ma->ma_valid & MA_INODE) && ma->ma_attr.la_valid) {
 501                 LASSERT((ma->ma_valid & MA_LOV) == 0);
 502                 rc = mdt_attr_set(info, mo, ma, rr->rr_flags);
 503                 if (rc)
 504                         GOTO(out_put, rc);
 505         } else if ((ma->ma_valid & MA_LOV) && (ma->ma_valid & MA_INODE)) {
 506                 struct lu_buf *buf  = &info->mti_buf;
 507                 LASSERT(ma->ma_attr.la_valid == 0);
 508                 buf->lb_buf = ma->ma_lmm;
 509                 buf->lb_len = ma->ma_lmm_size;
 510                 rc = mo_xattr_set(info->mti_env, mdt_object_child(mo),
 511                                   buf, XATTR_NAME_LOV, 0);
 512                 if (rc)
 513                         GOTO(out_put, rc);
 514         } else
 515                 LBUG();
 516
 517         ma->ma_need = MA_INODE;
 518         ma->ma_valid = 0;
 519         rc = mdt_attr_get_complex(info, mo, ma);
 520         if (rc != 0)
 521                 GOTO(out_put, rc);
 522
 523         mdt_pack_attr2body(info, repbody, &ma->ma_attr, mdt_object_fid(mo));
 524
 525         if (info->mti_mdt->mdt_opts.mo_oss_capa &&
 526             info->mti_exp->exp_connect_flags & OBD_CONNECT_OSS_CAPA &&
 527             S_ISREG(lu_object_attr(&mo->mot_obj.mo_lu)) &&
 528             (ma->ma_attr.la_valid & LA_SIZE) && !som_au) {
 529                 struct lustre_capa *capa;
 530
 531                 capa = req_capsule_server_get(info->mti_pill, &RMF_CAPA2);
 532                 LASSERT(capa);
 533                 capa->lc_opc = CAPA_OPC_OSS_DEFAULT | CAPA_OPC_OSS_TRUNC;
 534                 rc = mo_capa_get(info->mti_env, mdt_object_child(mo), capa, 0);
 535                 if (rc)
 536                         GOTO(out_put, rc);
 537                 repbody->valid |= OBD_MD_FLOSSCAPA;
 538         }
 539
 540         EXIT;
 541 out_put:
 542         mdt_object_put(info->mti_env, mo);
 543 out:
 544         if (rc == 0)
 545                 mdt_counter_incr(req, LPROC_MDT_SETATTR);
 546
 547         mdt_client_compatibility(info);
 548         rc2 = mdt_fix_reply(info);
 549         if (rc == 0)
 550                 rc = rc2;
 551         return rc;
 552 }
 553
 554 static int mdt_reint_create(struct mdt_thread_info *info,
 555                             struct mdt_lock_handle *lhc)
 556 {
 557         struct ptlrpc_request   *req = mdt_info_req(info);
 558         int                     rc;
 559         ENTRY;
 560
 561         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
 562                 RETURN(err_serious(-ESTALE));
 563
 564         if (info->mti_dlm_req)
 565                 ldlm_request_cancel(mdt_info_req(info), info->mti_dlm_req, 0);
 566
 567         LASSERT(info->mti_rr.rr_namelen > 0);
 568         switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
 569         case S_IFDIR:
 570                 mdt_counter_incr(req, LPROC_MDT_MKDIR);
 571                 break;
 572         case S_IFREG:
 573         case S_IFLNK:
 574         case S_IFCHR:
 575         case S_IFBLK:
 576         case S_IFIFO:
 577         case S_IFSOCK:
 578                 /* Special file should stay on the same node as parent. */
 579                 mdt_counter_incr(req, LPROC_MDT_MKNOD);
 580                 break;
 581         default:
 582                 CERROR("%s: Unsupported mode %o\n",
 583                        mdt2obd_dev(info->mti_mdt)->obd_name,
 584                        info->mti_attr.ma_attr.la_mode);
 585                 RETURN(err_serious(-EOPNOTSUPP));
 586         }
 587
 588         rc = mdt_md_create(info);
 589         RETURN(rc);
 590 }
 591
 592 /*
 593  * VBR: save parent version in reply and child version getting by its name.
 594  * Version of child is getting and checking during its lookup. If
 595  */
 596 static int mdt_reint_unlink(struct mdt_thread_info *info,
 597                             struct mdt_lock_handle *lhc)
 598 {
 599         struct mdt_reint_record *rr = &info->mti_rr;
 600         struct ptlrpc_request   *req = mdt_info_req(info);
 601         struct md_attr          *ma = &info->mti_attr;
 602         struct lu_fid           *child_fid = &info->mti_tmp_fid1;
 603         struct mdt_object       *mp;
 604         struct mdt_object       *mc;
 605         struct mdt_lock_handle  *parent_lh;
 606         struct mdt_lock_handle  *child_lh;
 607         struct lu_name          *lname;
 608         int                      rc;
 609         ENTRY;
 610
 611         DEBUG_REQ(D_INODE, req, "unlink "DFID"/%s", PFID(rr->rr_fid1),
 612                   rr->rr_name);
 613
 614         if (info->mti_dlm_req)
 615                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 616
 617         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
 618                 RETURN(err_serious(-ENOENT));
 619
 620         /*
 621          * step 1: lock the parent.
 622          */
 623         parent_lh = &info->mti_lh[MDT_LH_PARENT];
 624         mdt_lock_pdo_init(parent_lh, LCK_PW, rr->rr_name,
 625                           rr->rr_namelen);
 626
 627         mp = mdt_object_find_lock(info, rr->rr_fid1, parent_lh,
 628                                   MDS_INODELOCK_UPDATE);
 629         if (IS_ERR(mp))
 630                 GOTO(out, rc = PTR_ERR(mp));
 631
 632         if (mdt_object_obf(mp))
 633                 GOTO(out_unlock_parent, rc = -EPERM);
 634
 635         rc = mdt_version_get_check_save(info, mp, 0);
 636         if (rc)
 637                 GOTO(out_unlock_parent, rc);
 638
 639         mdt_reint_init_ma(info, ma);
 640
 641         /* step 2: find & lock the child */
 642         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 643         /* lookup child object along with version checking */
 644         fid_zero(child_fid);
 645         rc = mdt_lookup_version_check(info, mp, lname, child_fid, 1);
 646         if (rc != 0)
 647                  GOTO(out_unlock_parent, rc);
 648
 649         /* We will lock the child regardless it is local or remote. No harm. */
 650         mc = mdt_object_find(info->mti_env, info->mti_mdt, child_fid);
 651         if (IS_ERR(mc))
 652                 GOTO(out_unlock_parent, rc = PTR_ERR(mc));
 653         child_lh = &info->mti_lh[MDT_LH_CHILD];
 654         mdt_lock_reg_init(child_lh, LCK_EX);
 655         rc = mdt_object_lock(info, mc, child_lh, MDS_INODELOCK_FULL,
 656                              MDT_CROSS_LOCK);
 657         if (rc != 0) {
 658                 mdt_object_put(info->mti_env, mc);
 659                 GOTO(out_unlock_parent, rc);
 660         }
 661
 662         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 663                        OBD_FAIL_MDS_REINT_UNLINK_WRITE);
 664         /* save version when object is locked */
 665         mdt_version_get_save(info, mc, 1);
 666         /*
 667          * Now we can only make sure we need MA_INODE, in mdd layer, will check
 668          * whether need MA_LOV and MA_COOKIE.
 669          */
 670         ma->ma_need = MA_INODE;
 671         ma->ma_valid = 0;
 672         mdt_set_capainfo(info, 1, child_fid, BYPASS_CAPA);
 673         rc = mdo_unlink(info->mti_env, mdt_object_child(mp),
 674                         mdt_object_child(mc), lname, ma);
 675         if (rc == 0 && !lu_object_is_dying(&mc->mot_header))
 676                 rc = mdt_attr_get_complex(info, mc, ma);
 677         if (rc == 0)
 678                 mdt_handle_last_unlink(info, mc, ma);
 679
 680         if (ma->ma_valid & MA_INODE) {
 681                 switch (ma->ma_attr.la_mode & S_IFMT) {
 682                 case S_IFDIR:
 683                         mdt_counter_incr(req, LPROC_MDT_RMDIR);
 684                         break;
 685                 case S_IFREG:
 686                 case S_IFLNK:
 687                 case S_IFCHR:
 688                 case S_IFBLK:
 689                 case S_IFIFO:
 690                 case S_IFSOCK:
 691                         mdt_counter_incr(req, LPROC_MDT_UNLINK);
 692                         break;
 693                 default:
 694                         LASSERTF(0, "bad file type %o unlinking\n",
 695                                  ma->ma_attr.la_mode);
 696                 }
 697         }
 698
 699         EXIT;
 700
 701         mdt_object_unlock_put(info, mc, child_lh, rc);
 702 out_unlock_parent:
 703         mdt_object_unlock_put(info, mp, parent_lh, rc);
 704 out:
 705         return rc;
 706 }
 707
 708 /*
 709  * VBR: save versions in reply: 0 - parent; 1 - child by fid; 2 - target by
 710  * name.
 711  */
 712 static int mdt_reint_link(struct mdt_thread_info *info,
 713                           struct mdt_lock_handle *lhc)
 714 {
 715         struct mdt_reint_record *rr = &info->mti_rr;
 716         struct ptlrpc_request   *req = mdt_info_req(info);
 717         struct md_attr          *ma = &info->mti_attr;
 718         struct mdt_object       *ms;
 719         struct mdt_object       *mp;
 720         struct mdt_lock_handle  *lhs;
 721         struct mdt_lock_handle  *lhp;
 722         struct lu_name          *lname;
 723         int rc;
 724         ENTRY;
 725
 726         DEBUG_REQ(D_INODE, req, "link "DFID" to "DFID"/%s",
 727                   PFID(rr->rr_fid1), PFID(rr->rr_fid2), rr->rr_name);
 728
 729         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
 730                 RETURN(err_serious(-ENOENT));
 731
 732         if (info->mti_dlm_req)
 733                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 734
 735         /* Invalid case so return error immediately instead of
 736          * processing it */
 737         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2))
 738                 RETURN(-EPERM);
 739
 740         /* step 1: find & lock the target parent dir */
 741         lhp = &info->mti_lh[MDT_LH_PARENT];
 742         mdt_lock_pdo_init(lhp, LCK_PW, rr->rr_name,
 743                           rr->rr_namelen);
 744         mp = mdt_object_find_lock(info, rr->rr_fid2, lhp,
 745                                   MDS_INODELOCK_UPDATE);
 746         if (IS_ERR(mp))
 747                 RETURN(PTR_ERR(mp));
 748
 749         if (mdt_object_obf(mp))
 750                 GOTO(out_unlock_parent, rc = -EPERM);
 751
 752         rc = mdt_version_get_check_save(info, mp, 0);
 753         if (rc)
 754                 GOTO(out_unlock_parent, rc);
 755
 756         /* step 2: find & lock the source */
 757         lhs = &info->mti_lh[MDT_LH_CHILD];
 758         mdt_lock_reg_init(lhs, LCK_EX);
 759
 760         ms = mdt_object_find(info->mti_env, info->mti_mdt, rr->rr_fid1);
 761         if (IS_ERR(ms))
 762                 GOTO(out_unlock_parent, rc = PTR_ERR(ms));
 763
 764         rc = mdt_object_lock(info, ms, lhs, MDS_INODELOCK_UPDATE,
 765                             MDT_CROSS_LOCK);
 766         if (rc != 0) {
 767                 mdt_object_put(info->mti_env, ms);
 768                 GOTO(out_unlock_parent, rc);
 769         }
 770
 771         /* step 3: link it */
 772         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
 773                        OBD_FAIL_MDS_REINT_LINK_WRITE);
 774
 775         info->mti_mos = ms;
 776         rc = mdt_version_get_check_save(info, ms, 1);
 777         if (rc)
 778                 GOTO(out_unlock_child, rc);
 779
 780         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
 781         /** check target version by name during replay */
 782         rc = mdt_lookup_version_check(info, mp, lname, &info->mti_tmp_fid1, 2);
 783         if (rc != 0 && rc != -ENOENT)
 784                 GOTO(out_unlock_child, rc);
 785         /* save version of file name for replay, it must be ENOENT here */
 786         if (!req_is_replay(mdt_info_req(info))) {
 787                 info->mti_ver[2] = ENOENT_VERSION;
 788                 mdt_version_save(mdt_info_req(info), info->mti_ver[2], 2);
 789         }
 790
 791         rc = mdo_link(info->mti_env, mdt_object_child(mp),
 792                       mdt_object_child(ms), lname, ma);
 793
 794         if (rc == 0)
 795                 mdt_counter_incr(req, LPROC_MDT_LINK);
 796
 797         EXIT;
 798 out_unlock_child:
 799         mdt_object_unlock_put(info, ms, lhs, rc);
 800 out_unlock_parent:
 801         mdt_object_unlock_put(info, mp, lhp, rc);
 802         return rc;
 803 }
 804 /**
 805  * lock the part of the directory according to the hash of the name
 806  * (lh->mlh_pdo_hash) in parallel directory lock.
 807  */
 808 static int mdt_pdir_hash_lock(struct mdt_thread_info *info,
 809                        struct mdt_lock_handle *lh,
 810                        struct mdt_object *obj, __u64 ibits)
 811 {
 812         struct ldlm_res_id *res_id = &info->mti_res_id;
 813         struct ldlm_namespace *ns = info->mti_mdt->mdt_namespace;
 814         ldlm_policy_data_t *policy = &info->mti_policy;
 815         int rc;
 816
 817         /*
 818          * Finish res_id initializing by name hash marking part of
 819          * directory which is taking modification.
 820          */
 821         LASSERT(lh->mlh_pdo_hash != 0);
 822         fid_build_pdo_res_name(mdt_object_fid(obj), lh->mlh_pdo_hash, res_id);
 823         memset(policy, 0, sizeof(*policy));
 824         policy->l_inodebits.bits = ibits;
 825         /*
 826          * Use LDLM_FL_LOCAL_ONLY for this lock. We do not know yet if it is
 827          * going to be sent to client. If it is - mdt_intent_policy() path will
 828          * fix it up and turn FL_LOCAL flag off.
 829          */
 830         rc = mdt_fid_lock(ns, &lh->mlh_reg_lh, lh->mlh_reg_mode, policy,
 831                           res_id, LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB,
 832                           &info->mti_exp->exp_handle.h_cookie);
 833         return rc;
 834 }
 835
 836 static int mdt_rename_lock(struct mdt_thread_info *info,
 837                            struct lustre_handle *lh)
 838 {
 839         struct ldlm_namespace *ns     = info->mti_mdt->mdt_namespace;
 840         ldlm_policy_data_t    *policy = &info->mti_policy;
 841         struct ldlm_res_id    *res_id = &info->mti_res_id;
 842         __u64                  flags = 0;
 843         struct md_site        *ms;
 844         int rc;
 845         ENTRY;
 846
 847         ms = mdt_md_site(info->mti_mdt);
 848         fid_build_reg_res_name(&LUSTRE_BFL_FID, res_id);
 849
 850         memset(policy, 0, sizeof *policy);
 851         policy->l_inodebits.bits = MDS_INODELOCK_UPDATE;
 852
 853         if (ms->ms_control_exp == NULL) {
 854                 flags = LDLM_FL_LOCAL_ONLY | LDLM_FL_ATOMIC_CB;
 855
 856                 /*
 857                  * Current node is controller, that is mdt0, where we should
 858                  * take BFL lock.
 859                  */
 860                 rc = ldlm_cli_enqueue_local(ns, res_id, LDLM_IBITS, policy,
 861                                             LCK_EX, &flags, ldlm_blocking_ast,
 862                                             ldlm_completion_ast, NULL, NULL, 0,
 863                                             &info->mti_exp->exp_handle.h_cookie,
 864                                             lh);
 865         } else {
 866                 struct ldlm_enqueue_info einfo = { LDLM_IBITS, LCK_EX,
 867                      ldlm_blocking_ast, ldlm_completion_ast, NULL, NULL, NULL };
 868                 /*
 869                  * This is the case mdt0 is remote node, issue DLM lock like
 870                  * other clients.
 871                  */
 872                 rc = ldlm_cli_enqueue(ms->ms_control_exp, NULL, &einfo, res_id,
 873                                       policy, &flags, NULL, 0, lh, 0);
 874         }
 875
 876         RETURN(rc);
 877 }
 878
 879 static void mdt_rename_unlock(struct lustre_handle *lh)
 880 {
 881         ENTRY;
 882         LASSERT(lustre_handle_is_used(lh));
 883         ldlm_lock_decref(lh, LCK_EX);
 884         EXIT;
 885 }
 886
 887 /*
 888  * This is is_subdir() variant, it is CMD if cmm forwards it to correct
 889  * target. Source should not be ancestor of target dir. May be other rename
 890  * checks can be moved here later.
 891  */
 892 static int mdt_rename_sanity(struct mdt_thread_info *info, struct lu_fid *fid)
 893 {
 894         struct mdt_reint_record *rr = &info->mti_rr;
 895         struct lu_fid dst_fid = *rr->rr_fid2;
 896         struct mdt_object *dst;
 897         int rc = 0;
 898         ENTRY;
 899
 900         do {
 901                 LASSERT(fid_is_sane(&dst_fid));
 902                 dst = mdt_object_find(info->mti_env, info->mti_mdt, &dst_fid);
 903                 if (!IS_ERR(dst)) {
 904                         rc = mdo_is_subdir(info->mti_env,
 905                                            mdt_object_child(dst), fid,
 906                                            &dst_fid);
 907                         mdt_object_put(info->mti_env, dst);
 908                         if (rc != -EREMOTE && rc < 0) {
 909                                 CERROR("Failed mdo_is_subdir(), rc %d\n", rc);
 910                         } else {
 911                                 /* check the found fid */
 912                                 if (lu_fid_eq(&dst_fid, fid))
 913                                         rc = -EINVAL;
 914                         }
 915                 } else {
 916                         rc = PTR_ERR(dst);
 917                 }
 918         } while (rc == -EREMOTE);
 919
 920         RETURN(rc);
 921 }
 922
 923 /*
 924  * VBR: rename versions in reply: 0 - src parent; 1 - tgt parent;
 925  * 2 - src child; 3 - tgt child.
 926  * Update on disk version of src child.
 927  */
 928 static int mdt_reint_rename(struct mdt_thread_info *info,
 929                             struct mdt_lock_handle *lhc)
 930 {
 931         struct mdt_reint_record *rr = &info->mti_rr;
 932         struct md_attr          *ma = &info->mti_attr;
 933         struct ptlrpc_request   *req = mdt_info_req(info);
 934         struct mdt_object       *msrcdir;
 935         struct mdt_object       *mtgtdir;
 936         struct mdt_object       *mold;
 937         struct mdt_object       *mnew = NULL;
 938         struct mdt_lock_handle  *lh_srcdirp;
 939         struct mdt_lock_handle  *lh_tgtdirp;
 940         struct mdt_lock_handle  *lh_oldp;
 941         struct mdt_lock_handle  *lh_newp;
 942         struct lu_fid           *old_fid = &info->mti_tmp_fid1;
 943         struct lu_fid           *new_fid = &info->mti_tmp_fid2;
 944         struct lustre_handle     rename_lh = { 0 };
 945         struct lu_name           slname = { 0 };
 946         struct lu_name          *lname;
 947         int                      rc;
 948         ENTRY;
 949
 950         if (info->mti_dlm_req)
 951                 ldlm_request_cancel(req, info->mti_dlm_req, 0);
 952
 953         DEBUG_REQ(D_INODE, req, "rename "DFID"/%s to "DFID"/%s",
 954                   PFID(rr->rr_fid1), rr->rr_name,
 955                   PFID(rr->rr_fid2), rr->rr_tgt);
 956
 957         rc = mdt_rename_lock(info, &rename_lh);
 958         if (rc) {
 959                 CERROR("Can't lock FS for rename, rc %d\n", rc);
 960                 RETURN(rc);
 961         }
 962
 963         lh_newp = &info->mti_lh[MDT_LH_NEW];
 964
 965         /* step 1: lock the source dir. */
 966         lh_srcdirp = &info->mti_lh[MDT_LH_PARENT];
 967         mdt_lock_pdo_init(lh_srcdirp, LCK_PW, rr->rr_name,
 968                           rr->rr_namelen);
 969         msrcdir = mdt_object_find_lock(info, rr->rr_fid1, lh_srcdirp,
 970                                        MDS_INODELOCK_UPDATE);
 971         if (IS_ERR(msrcdir))
 972                 GOTO(out_rename_lock, rc = PTR_ERR(msrcdir));
 973
 974         if (mdt_object_obf(msrcdir))
 975                 GOTO(out_unlock_source, rc = -EPERM);
 976
 977         rc = mdt_version_get_check_save(info, msrcdir, 0);
 978         if (rc)
 979                 GOTO(out_unlock_source, rc);
 980
 981         /* step 2: find & lock the target dir. */
 982         lh_tgtdirp = &info->mti_lh[MDT_LH_CHILD];
 983         mdt_lock_pdo_init(lh_tgtdirp, LCK_PW, rr->rr_tgt,
 984                           rr->rr_tgtlen);
 985         if (lu_fid_eq(rr->rr_fid1, rr->rr_fid2)) {
 986                 mdt_object_get(info->mti_env, msrcdir);
 987                 mtgtdir = msrcdir;
 988                 if (lh_tgtdirp->mlh_pdo_hash != lh_srcdirp->mlh_pdo_hash) {
 989                          rc = mdt_pdir_hash_lock(info, lh_tgtdirp, mtgtdir,
 990                                                  MDS_INODELOCK_UPDATE);
 991                          if (rc)
 992                                  GOTO(out_unlock_source, rc);
 993                          OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PDO_LOCK2, 10);
 994                 }
 995         } else {
 996                 mtgtdir = mdt_object_find(info->mti_env, info->mti_mdt,
 997                                           rr->rr_fid2);
 998                 if (IS_ERR(mtgtdir))
 999                         GOTO(out_unlock_source, rc = PTR_ERR(mtgtdir));
1000
1001                 if (mdt_object_obf(mtgtdir))
1002                         GOTO(out_put_target, rc = -EPERM);
1003
1004                 /* check early, the real version will be saved after locking */
1005                 rc = mdt_version_get_check(info, mtgtdir, 1);
1006                 if (rc)
1007                         GOTO(out_put_target, rc);
1008
1009                 rc = mdt_object_exists(mtgtdir);
1010                 if (rc == 0) {
1011                         GOTO(out_put_target, rc = -ESTALE);
1012                 } else if (rc > 0) {
1013                         /* we lock the target dir if it is local */
1014                         rc = mdt_object_lock(info, mtgtdir, lh_tgtdirp,
1015                                              MDS_INODELOCK_UPDATE,
1016                                              MDT_LOCAL_LOCK);
1017                         if (rc != 0)
1018                                 GOTO(out_put_target, rc);
1019                         /* get and save correct version after locking */
1020                         mdt_version_get_save(info, mtgtdir, 1);
1021                 }
1022         }
1023
1024         /* step 3: find & lock the old object. */
1025         lname = mdt_name(info->mti_env, (char *)rr->rr_name, rr->rr_namelen);
1026         mdt_name_copy(&slname, lname);
1027         fid_zero(old_fid);
1028         rc = mdt_lookup_version_check(info, msrcdir, &slname, old_fid, 2);
1029         if (rc != 0)
1030                 GOTO(out_unlock_target, rc);
1031
1032         if (lu_fid_eq(old_fid, rr->rr_fid1) || lu_fid_eq(old_fid, rr->rr_fid2))
1033                 GOTO(out_unlock_target, rc = -EINVAL);
1034
1035         mold = mdt_object_find(info->mti_env, info->mti_mdt, old_fid);
1036         if (IS_ERR(mold))
1037                 GOTO(out_unlock_target, rc = PTR_ERR(mold));
1038
1039         if (mdt_object_obf(mold)) {
1040                 mdt_object_put(info->mti_env, mold);
1041                 GOTO(out_unlock_target, rc = -EPERM);
1042         }
1043
1044         lh_oldp = &info->mti_lh[MDT_LH_OLD];
1045         mdt_lock_reg_init(lh_oldp, LCK_EX);
1046         rc = mdt_object_lock(info, mold, lh_oldp, MDS_INODELOCK_LOOKUP,
1047                              MDT_CROSS_LOCK);
1048         if (rc != 0) {
1049                 mdt_object_put(info->mti_env, mold);
1050                 GOTO(out_unlock_target, rc);
1051         }
1052
1053         info->mti_mos = mold;
1054         /* save version after locking */
1055         mdt_version_get_save(info, mold, 2);
1056         mdt_set_capainfo(info, 2, old_fid, BYPASS_CAPA);
1057
1058         /* step 4: find & lock the new object. */
1059         /* new target object may not exist now */
1060         lname = mdt_name(info->mti_env, (char *)rr->rr_tgt, rr->rr_tgtlen);
1061         /* lookup with version checking */
1062         fid_zero(new_fid);
1063         rc = mdt_lookup_version_check(info, mtgtdir, lname, new_fid, 3);
1064         if (rc == 0) {
1065                 /* the new_fid should have been filled at this moment */
1066                 if (lu_fid_eq(old_fid, new_fid))
1067                        GOTO(out_unlock_old, rc);
1068
1069                 if (lu_fid_eq(new_fid, rr->rr_fid1) ||
1070                     lu_fid_eq(new_fid, rr->rr_fid2))
1071                         GOTO(out_unlock_old, rc = -EINVAL);
1072
1073                 mdt_lock_reg_init(lh_newp, LCK_EX);
1074                 mnew = mdt_object_find(info->mti_env, info->mti_mdt, new_fid);
1075                 if (IS_ERR(mnew))
1076                         GOTO(out_unlock_old, rc = PTR_ERR(mnew));
1077
1078                 if (mdt_object_obf(mnew)) {
1079                         mdt_object_put(info->mti_env, mnew);
1080                         GOTO(out_unlock_old, rc = -EPERM);
1081                 }
1082
1083                 rc = mdt_object_lock(info, mnew, lh_newp,
1084                                      MDS_INODELOCK_FULL, MDT_CROSS_LOCK);
1085                 if (rc != 0) {
1086                         mdt_object_put(info->mti_env, mnew);
1087                         GOTO(out_unlock_old, rc);
1088                 }
1089                 /* get and save version after locking */
1090                 mdt_version_get_save(info, mnew, 3);
1091                 mdt_set_capainfo(info, 3, new_fid, BYPASS_CAPA);
1092         } else if (rc != -EREMOTE && rc != -ENOENT) {
1093                 GOTO(out_unlock_old, rc);
1094         } else {
1095                 mdt_enoent_version_save(info, 3);
1096         }
1097
1098         /* step 5: rename it */
1099         mdt_reint_init_ma(info, ma);
1100
1101         mdt_fail_write(info->mti_env, info->mti_mdt->mdt_bottom,
1102                        OBD_FAIL_MDS_REINT_RENAME_WRITE);
1103
1104
1105         /* Check if @dst is subdir of @src. */
1106         rc = mdt_rename_sanity(info, old_fid);
1107         if (rc)
1108                 GOTO(out_unlock_new, rc);
1109
1110         rc = mdo_rename(info->mti_env, mdt_object_child(msrcdir),
1111                         mdt_object_child(mtgtdir), old_fid, &slname,
1112                         (mnew ? mdt_object_child(mnew) : NULL),
1113                         lname, ma);
1114
1115         /* handle last link of tgt object */
1116         if (rc == 0) {
1117                 mdt_counter_incr(req, LPROC_MDT_RENAME);
1118                 if (mnew)
1119                         mdt_handle_last_unlink(info, mnew, ma);
1120
1121                 mdt_rename_counter_tally(info, info->mti_mdt, req,
1122                                          msrcdir, mtgtdir);
1123         }
1124
1125         EXIT;
1126 out_unlock_new:
1127         if (mnew)
1128                 mdt_object_unlock_put(info, mnew, lh_newp, rc);
1129 out_unlock_old:
1130         mdt_object_unlock_put(info, mold, lh_oldp, rc);
1131 out_unlock_target:
1132         mdt_object_unlock(info, mtgtdir, lh_tgtdirp, rc);
1133 out_put_target:
1134         mdt_object_put(info->mti_env, mtgtdir);
1135 out_unlock_source:
1136         mdt_object_unlock_put(info, msrcdir, lh_srcdirp, rc);
1137 out_rename_lock:
1138         mdt_rename_unlock(&rename_lh);
1139         return rc;
1140 }
1141
1142 typedef int (*mdt_reinter)(struct mdt_thread_info *info,
1143                            struct mdt_lock_handle *lhc);
1144
1145 static mdt_reinter reinters[REINT_MAX] = {
1146         [REINT_SETATTR]  = mdt_reint_setattr,
1147         [REINT_CREATE]   = mdt_reint_create,
1148         [REINT_LINK]     = mdt_reint_link,
1149         [REINT_UNLINK]   = mdt_reint_unlink,
1150         [REINT_RENAME]   = mdt_reint_rename,
1151         [REINT_OPEN]     = mdt_reint_open,
1152         [REINT_SETXATTR] = mdt_reint_setxattr
1153 };
1154
1155 int mdt_reint_rec(struct mdt_thread_info *info,
1156                   struct mdt_lock_handle *lhc)
1157 {
1158         int rc;
1159         ENTRY;
1160
1161         rc = reinters[info->mti_rr.rr_opcode](info, lhc);
1162
1163         RETURN(rc);
1164 }