lustre/mdd/mdd_dir.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.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2016, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/mdd/mdd_dir.c
  33  *
  34  * Lustre Metadata Server (mdd) routines
  35  *
  36  * Author: Wang Di <wangdi@intel.com>
  37  */
  38
  39 #define DEBUG_SUBSYSTEM S_MDS
  40
  41 #include <obd_class.h>
  42 #include <obd_support.h>
  43 #include <lustre_mds.h>
  44 #include <lustre_fid.h>
  45
  46 #include "mdd_internal.h"
  47
  48 static const char dot[] = ".";
  49 static const char dotdot[] = "..";
  50
  51 static struct lu_name lname_dotdot = {
  52         (char *) dotdot,
  53         sizeof(dotdot) - 1
  54 };
  55
  56 static inline int
  57 mdd_name_check(struct mdd_device *m, const struct lu_name *ln)
  58 {
  59         if (!lu_name_is_valid(ln))
  60                 return -EINVAL;
  61         else if (ln->ln_namelen > m->mdd_dt_conf.ddp_max_name_len)
  62                 return -ENAMETOOLONG;
  63         else
  64                 return 0;
  65 }
  66
  67 /* Get FID from name and parent */
  68 static int
  69 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
  70              const struct lu_attr *pattr, const struct lu_name *lname,
  71              struct lu_fid* fid, int mask)
  72 {
  73         const char *name                = lname->ln_name;
  74         const struct dt_key *key        = (const struct dt_key *)name;
  75         struct mdd_object *mdd_obj      = md2mdd_obj(pobj);
  76         struct mdd_device *m            = mdo2mdd(pobj);
  77         struct dt_object *dir           = mdd_object_child(mdd_obj);
  78         int rc;
  79         ENTRY;
  80
  81         if (unlikely(mdd_is_dead_obj(mdd_obj)))
  82                 RETURN(-ESTALE);
  83
  84         if (!mdd_object_exists(mdd_obj))
  85                 RETURN(-ESTALE);
  86
  87         if (mdd_object_remote(mdd_obj)) {
  88                 CDEBUG(D_INFO, "%s: Object "DFID" locates on remote server\n",
  89                        mdd2obd_dev(m)->obd_name, PFID(mdo2fid(mdd_obj)));
  90         }
  91
  92         rc = mdd_permission_internal_locked(env, mdd_obj, pattr, mask,
  93                                             MOR_TGT_PARENT);
  94         if (rc)
  95                 RETURN(rc);
  96
  97         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
  98                    dt_try_as_dir(env, dir)))
  99                 rc = dt_lookup(env, dir, (struct dt_rec *)fid, key);
 100         else
 101                 rc = -ENOTDIR;
 102
 103         RETURN(rc);
 104 }
 105
 106 int mdd_lookup(const struct lu_env *env,
 107                struct md_object *pobj, const struct lu_name *lname,
 108                struct lu_fid *fid, struct md_op_spec *spec)
 109 {
 110         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
 111         int rc;
 112         ENTRY;
 113
 114         rc = mdd_la_get(env, md2mdd_obj(pobj), pattr);
 115         if (rc != 0)
 116                 RETURN(rc);
 117
 118         rc = __mdd_lookup(env, pobj, pattr, lname, fid,
 119                           (spec != NULL && spec->sp_permitted) ? 0 : MAY_EXEC);
 120         RETURN(rc);
 121 }
 122
 123 /** Read the link EA into a temp buffer.
 124  * Uses the mdd_thread_info::mti_big_buf since it is generally large.
 125  * A pointer to the buffer is stored in \a ldata::ld_buf.
 126  *
 127  * \retval 0 or error
 128  */
 129 static int __mdd_links_read(const struct lu_env *env,
 130                             struct mdd_object *mdd_obj,
 131                             struct linkea_data *ldata)
 132 {
 133         int rc;
 134
 135         if (!mdd_object_exists(mdd_obj))
 136                 return -ENODATA;
 137
 138         /* First try a small buf */
 139         LASSERT(env != NULL);
 140         ldata->ld_buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_link_buf,
 141                                                PAGE_SIZE);
 142         if (ldata->ld_buf->lb_buf == NULL)
 143                 return -ENOMEM;
 144
 145         rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf, XATTR_NAME_LINK);
 146         if (rc == -ERANGE) {
 147                 /* Buf was too small, figure out what we need. */
 148                 lu_buf_free(ldata->ld_buf);
 149                 rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf,
 150                                    XATTR_NAME_LINK);
 151                 if (rc < 0)
 152                         return rc;
 153                 ldata->ld_buf = lu_buf_check_and_alloc(ldata->ld_buf, rc);
 154                 if (ldata->ld_buf->lb_buf == NULL)
 155                         return -ENOMEM;
 156                 rc = mdo_xattr_get(env, mdd_obj, ldata->ld_buf,
 157                                   XATTR_NAME_LINK);
 158         }
 159         if (rc < 0) {
 160                 lu_buf_free(ldata->ld_buf);
 161                 ldata->ld_buf = NULL;
 162                 return rc;
 163         }
 164
 165         return linkea_init(ldata);
 166 }
 167
 168 static int mdd_links_read(const struct lu_env *env,
 169                           struct mdd_object *mdd_obj,
 170                           struct linkea_data *ldata)
 171 {
 172         int rc;
 173
 174         rc = __mdd_links_read(env, mdd_obj, ldata);
 175         if (!rc)
 176                 rc = linkea_init(ldata);
 177
 178         return rc;
 179 }
 180
 181 static int mdd_links_read_with_rec(const struct lu_env *env,
 182                                    struct mdd_object *mdd_obj,
 183                                    struct linkea_data *ldata)
 184 {
 185         int rc;
 186
 187         rc = __mdd_links_read(env, mdd_obj, ldata);
 188         if (!rc)
 189                 rc = linkea_init_with_rec(ldata);
 190
 191         return rc;
 192 }
 193
 194 /**
 195  * Get parent FID of the directory
 196  *
 197  * Read parent FID from linkEA, if that fails, then do lookup
 198  * dotdot to get the parent FID.
 199  *
 200  * \param[in] env       execution environment
 201  * \param[in] obj       object from which to find the parent FID
 202  * \param[in] attr      attribute of the object
 203  * \param[out] fid      fid to get the parent FID
 204  *
 205  * \retval              0 if getting the parent FID succeeds.
 206  * \retval              negative errno if getting the parent FID fails.
 207  **/
 208 static inline int mdd_parent_fid(const struct lu_env *env,
 209                                  struct mdd_object *obj,
 210                                  const struct lu_attr *attr,
 211                                  struct lu_fid *fid)
 212 {
 213         struct mdd_thread_info  *info = mdd_env_info(env);
 214         struct linkea_data      ldata = { NULL };
 215         struct lu_buf           *buf = &info->mti_link_buf;
 216         struct lu_name          lname;
 217         int                     rc = 0;
 218
 219         ENTRY;
 220
 221         LASSERT(S_ISDIR(mdd_object_type(obj)));
 222
 223         buf = lu_buf_check_and_alloc(buf, PATH_MAX);
 224         if (buf->lb_buf == NULL)
 225                 GOTO(lookup, rc = 0);
 226
 227         ldata.ld_buf = buf;
 228         rc = mdd_links_read_with_rec(env, obj, &ldata);
 229         if (rc != 0)
 230                 GOTO(lookup, rc);
 231
 232         LASSERT(ldata.ld_leh != NULL);
 233         /* Directory should only have 1 parent */
 234         if (ldata.ld_leh->leh_reccount > 1)
 235                 GOTO(lookup, rc);
 236
 237         ldata.ld_lee = (struct link_ea_entry *)(ldata.ld_leh + 1);
 238
 239         linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen, &lname, fid);
 240         if (likely(fid_is_sane(fid)))
 241                 RETURN(0);
 242 lookup:
 243         rc =  __mdd_lookup(env, &obj->mod_obj, attr, &lname_dotdot, fid, 0);
 244         RETURN(rc);
 245 }
 246
 247 /*
 248  * For root fid use special function, which does not compare version component
 249  * of fid. Version component is different for root fids on all MDTs.
 250  */
 251 int mdd_is_root(struct mdd_device *mdd, const struct lu_fid *fid)
 252 {
 253         return fid_seq(&mdd->mdd_root_fid) == fid_seq(fid) &&
 254                 fid_oid(&mdd->mdd_root_fid) == fid_oid(fid);
 255 }
 256
 257 /*
 258  * return 1: if lf is the fid of the ancestor of p1;
 259  * return 0: if not;
 260  *
 261  * return -EREMOTE: if remote object is found, in this
 262  * case fid of remote object is saved to @pf;
 263  *
 264  * otherwise: values < 0, errors.
 265  */
 266 static int mdd_is_parent(const struct lu_env *env,
 267                         struct mdd_device *mdd,
 268                         struct mdd_object *p1,
 269                         const struct lu_attr *attr,
 270                         const struct lu_fid *lf,
 271                         struct lu_fid *pf)
 272 {
 273         struct mdd_object *parent = NULL;
 274         struct lu_fid *pfid;
 275         int rc;
 276         ENTRY;
 277
 278         LASSERT(!lu_fid_eq(mdo2fid(p1), lf));
 279         pfid = &mdd_env_info(env)->mti_fid;
 280
 281         /* Check for root first. */
 282         if (mdd_is_root(mdd, mdo2fid(p1)))
 283                 RETURN(0);
 284
 285         for(;;) {
 286                 /* this is done recursively */
 287                 rc = mdd_parent_fid(env, p1, attr, pfid);
 288                 if (rc)
 289                         GOTO(out, rc);
 290                 if (mdd_is_root(mdd, pfid))
 291                         GOTO(out, rc = 0);
 292                 if (lu_fid_eq(pfid, &mdd->mdd_local_root_fid))
 293                         GOTO(out, rc = 0);
 294                 if (lu_fid_eq(pfid, lf))
 295                         GOTO(out, rc = 1);
 296                 if (parent != NULL)
 297                         mdd_object_put(env, parent);
 298
 299                 parent = mdd_object_find(env, mdd, pfid);
 300                 if (IS_ERR(parent))
 301                         GOTO(out, rc = PTR_ERR(parent));
 302
 303                 if (!mdd_object_exists(parent))
 304                         GOTO(out, rc = -EINVAL);
 305
 306                 p1 = parent;
 307         }
 308         EXIT;
 309 out:
 310         if (parent && !IS_ERR(parent))
 311                 mdd_object_put(env, parent);
 312         return rc;
 313 }
 314
 315 /*
 316  * No permission check is needed.
 317  *
 318  * returns 1: if fid is ancestor of @mo;
 319  * returns 0: if fid is not an ancestor of @mo;
 320  *
 321  * returns EREMOTE if remote object is found, fid of remote object is saved to
 322  * @fid;
 323  *
 324  * returns < 0: if error
 325  */
 326 int mdd_is_subdir(const struct lu_env *env, struct md_object *mo,
 327                   const struct lu_fid *fid, struct lu_fid *sfid)
 328 {
 329         struct mdd_device *mdd = mdo2mdd(mo);
 330         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
 331         int rc;
 332         ENTRY;
 333
 334         if (!S_ISDIR(mdd_object_type(md2mdd_obj(mo))))
 335                 RETURN(0);
 336
 337         rc = mdd_la_get(env, md2mdd_obj(mo), attr);
 338         if (rc != 0)
 339                 RETURN(rc);
 340
 341         rc = mdd_is_parent(env, mdd, md2mdd_obj(mo), attr, fid, sfid);
 342         if (rc == 0) {
 343                 /* found root */
 344                 fid_zero(sfid);
 345         } else if (rc == 1) {
 346                 /* found @fid is parent */
 347                 *sfid = *fid;
 348                 rc = 0;
 349         }
 350         RETURN(rc);
 351 }
 352
 353 /*
 354  * Check that @dir contains no entries except (possibly) dot and dotdot.
 355  *
 356  * Returns:
 357  *
 358  *             0        empty
 359  *      -ENOTDIR        not a directory object
 360  *    -ENOTEMPTY        not empty
 361  *           -ve        other error
 362  *
 363  */
 364 static int mdd_dir_is_empty(const struct lu_env *env,
 365                             struct mdd_object *dir)
 366 {
 367         struct dt_it     *it;
 368         struct dt_object *obj;
 369         const struct dt_it_ops *iops;
 370         int result;
 371         ENTRY;
 372
 373         obj = mdd_object_child(dir);
 374         if (!dt_try_as_dir(env, obj))
 375                 RETURN(-ENOTDIR);
 376
 377         iops = &obj->do_index_ops->dio_it;
 378         it = iops->init(env, obj, LUDA_64BITHASH);
 379         if (!IS_ERR(it)) {
 380                 result = iops->get(env, it, (const struct dt_key *)"");
 381                 if (result > 0) {
 382                         int i;
 383                         for (result = 0, i = 0; result == 0 && i < 3; ++i)
 384                                 result = iops->next(env, it);
 385                         if (result == 0)
 386                                 result = -ENOTEMPTY;
 387                         else if (result == 1)
 388                                 result = 0;
 389                 } else if (result == 0)
 390                         /*
 391                          * Huh? Index contains no zero key?
 392                          */
 393                         result = -EIO;
 394
 395                 iops->put(env, it);
 396                 iops->fini(env, it);
 397         } else
 398                 result = PTR_ERR(it);
 399         RETURN(result);
 400 }
 401
 402 /**
 403  * Determine if the target object can be hard linked, and right now it only
 404  * checks if the link count reach the maximum limit. Note: for ldiskfs, the
 405  * directory nlink count might exceed the maximum link count(see
 406  * osd_object_ref_add), so it only check nlink for non-directories.
 407  *
 408  * \param[in] env       thread environment
 409  * \param[in] obj       object being linked to
 410  * \param[in] la        attributes of \a obj
 411  *
 412  * \retval              0 if \a obj can be hard linked
 413  * \retval              negative error if \a obj is a directory or has too
 414  *                      many links
 415  */
 416 static int __mdd_may_link(const struct lu_env *env, struct mdd_object *obj,
 417                           const struct lu_attr *la)
 418 {
 419         struct mdd_device *m = mdd_obj2mdd_dev(obj);
 420         ENTRY;
 421
 422         LASSERT(la != NULL);
 423
 424         /* Subdir count limitation can be broken through
 425          * (see osd_object_ref_add), so only check non-directory here. */
 426         if (!S_ISDIR(la->la_mode) &&
 427             la->la_nlink >= m->mdd_dt_conf.ddp_max_nlink)
 428                 RETURN(-EMLINK);
 429
 430         RETURN(0);
 431 }
 432
 433 /**
 434  * Check whether it may create the cobj under the pobj.
 435  *
 436  * \param[in] env       execution environment
 437  * \param[in] pobj      the parent directory
 438  * \param[in] pattr     the attribute of the parent directory
 439  * \param[in] cobj      the child to be created
 440  * \param[in] check_perm        if check WRITE|EXEC permission for parent
 441  *
 442  * \retval              = 0 create the child under this dir is allowed
 443  * \retval              negative errno create the child under this dir is
 444  *                      not allowed
 445  */
 446 int mdd_may_create(const struct lu_env *env, struct mdd_object *pobj,
 447                    const struct lu_attr *pattr, struct mdd_object *cobj,
 448                    bool check_perm)
 449 {
 450         int rc = 0;
 451         ENTRY;
 452
 453         if (cobj && mdd_object_exists(cobj))
 454                 RETURN(-EEXIST);
 455
 456         if (mdd_is_dead_obj(pobj))
 457                 RETURN(-ENOENT);
 458
 459         if (check_perm)
 460                 rc = mdd_permission_internal_locked(env, pobj, pattr,
 461                                                     MAY_WRITE | MAY_EXEC,
 462                                                     MOR_TGT_PARENT);
 463         RETURN(rc);
 464 }
 465
 466 /*
 467  * Check whether can unlink from the pobj in the case of "cobj == NULL".
 468  */
 469 int mdd_may_unlink(const struct lu_env *env, struct mdd_object *pobj,
 470                    const struct lu_attr *pattr, const struct lu_attr *attr)
 471 {
 472         int rc;
 473         ENTRY;
 474
 475         if (mdd_is_dead_obj(pobj))
 476                 RETURN(-ENOENT);
 477
 478         if (attr->la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL))
 479                 RETURN(-EPERM);
 480
 481         rc = mdd_permission_internal_locked(env, pobj, pattr,
 482                                             MAY_WRITE | MAY_EXEC,
 483                                             MOR_TGT_PARENT);
 484         if (rc != 0)
 485                 RETURN(rc);
 486
 487         if (pattr->la_flags & LUSTRE_APPEND_FL)
 488                 RETURN(-EPERM);
 489
 490         RETURN(rc);
 491 }
 492
 493 /*
 494  * pobj == NULL is remote ops case, under such case, pobj's
 495  * VTX feature has been checked already, no need check again.
 496  */
 497 static inline int mdd_is_sticky(const struct lu_env *env,
 498                                 struct mdd_object *pobj,
 499                                 const struct lu_attr *pattr,
 500                                 struct mdd_object *cobj,
 501                                 const struct lu_attr *cattr)
 502 {
 503         struct lu_ucred *uc = lu_ucred_assert(env);
 504
 505         if (pobj != NULL) {
 506                 LASSERT(pattr != NULL);
 507                 if (!(pattr->la_mode & S_ISVTX) ||
 508                     (pattr->la_uid == uc->uc_fsuid))
 509                         return 0;
 510         }
 511
 512         LASSERT(cattr != NULL);
 513         if (cattr->la_uid == uc->uc_fsuid)
 514                 return 0;
 515
 516         return !md_capable(uc, CFS_CAP_FOWNER);
 517 }
 518
 519 static int mdd_may_delete_entry(const struct lu_env *env,
 520                                 struct mdd_object *pobj,
 521                                 const struct lu_attr *pattr,
 522                                 int check_perm)
 523 {
 524         ENTRY;
 525
 526         LASSERT(pobj != NULL);
 527         if (!mdd_object_exists(pobj))
 528                 RETURN(-ENOENT);
 529
 530         if (mdd_is_dead_obj(pobj))
 531                 RETURN(-ENOENT);
 532
 533         if (check_perm) {
 534                 int rc;
 535                 rc = mdd_permission_internal_locked(env, pobj, pattr,
 536                                             MAY_WRITE | MAY_EXEC,
 537                                             MOR_TGT_PARENT);
 538                 if (rc)
 539                         RETURN(rc);
 540         }
 541
 542         if (pattr->la_flags & LUSTRE_APPEND_FL)
 543                 RETURN(-EPERM);
 544
 545         RETURN(0);
 546 }
 547
 548 /*
 549  * Check whether it may delete the cobj from the pobj.
 550  * pobj maybe NULL
 551  */
 552 int mdd_may_delete(const struct lu_env *env, struct mdd_object *tpobj,
 553                    const struct lu_attr *tpattr, struct mdd_object *tobj,
 554                    const struct lu_attr *tattr, const struct lu_attr *cattr,
 555                    int check_perm, int check_empty)
 556 {
 557         int rc = 0;
 558         ENTRY;
 559
 560         if (tpobj) {
 561                 LASSERT(tpattr != NULL);
 562                 rc = mdd_may_delete_entry(env, tpobj, tpattr, check_perm);
 563                 if (rc != 0)
 564                         RETURN(rc);
 565         }
 566
 567         if (tobj == NULL)
 568                 RETURN(0);
 569
 570         if (!mdd_object_exists(tobj))
 571                 RETURN(-ENOENT);
 572
 573         if (mdd_is_dead_obj(tobj))
 574                 RETURN(-ESTALE);
 575
 576         if (mdd_is_sticky(env, tpobj, tpattr, tobj, tattr))
 577                 RETURN(-EPERM);
 578
 579         if (tattr->la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL))
 580                 RETURN(-EPERM);
 581
 582         /* additional check the rename case */
 583         if (cattr) {
 584                 if (S_ISDIR(cattr->la_mode)) {
 585                         struct mdd_device *mdd = mdo2mdd(&tobj->mod_obj);
 586
 587                         if (!S_ISDIR(tattr->la_mode))
 588                                 RETURN(-ENOTDIR);
 589
 590                         if (lu_fid_eq(mdo2fid(tobj), &mdd->mdd_root_fid))
 591                                 RETURN(-EBUSY);
 592                 } else if (S_ISDIR(tattr->la_mode))
 593                         RETURN(-EISDIR);
 594         }
 595
 596         if (S_ISDIR(tattr->la_mode) && check_empty)
 597                 rc = mdd_dir_is_empty(env, tobj);
 598
 599         RETURN(rc);
 600 }
 601
 602 /**
 603  * Check whether it can create the link file(linked to @src_obj) under
 604  * the target directory(@tgt_obj), and src_obj has been locked by
 605  * mdd_write_lock.
 606  *
 607  * \param[in] env       execution environment
 608  * \param[in] tgt_obj   the target directory
 609  * \param[in] tattr     attributes of target directory
 610  * \param[in] lname     the link name
 611  * \param[in] src_obj   source object for link
 612  * \param[in] cattr     attributes for source object
 613  *
 614  * \retval              = 0 it is allowed to create the link file under tgt_obj
 615  * \retval              negative error not allowed to create the link file
 616  */
 617 static int mdd_link_sanity_check(const struct lu_env *env,
 618                                  struct mdd_object *tgt_obj,
 619                                  const struct lu_attr *tattr,
 620                                  const struct lu_name *lname,
 621                                  struct mdd_object *src_obj,
 622                                  const struct lu_attr *cattr)
 623 {
 624         struct mdd_device *m = mdd_obj2mdd_dev(src_obj);
 625         int rc = 0;
 626         ENTRY;
 627
 628         if (!mdd_object_exists(src_obj))
 629                 RETURN(-ENOENT);
 630
 631         if (mdd_is_dead_obj(src_obj))
 632                 RETURN(-ESTALE);
 633
 634         /* Local ops, no lookup before link, check filename length here. */
 635         rc = mdd_name_check(m, lname);
 636         if (rc < 0)
 637                 RETURN(rc);
 638
 639         if (cattr->la_flags & (LUSTRE_IMMUTABLE_FL | LUSTRE_APPEND_FL))
 640                 RETURN(-EPERM);
 641
 642         if (S_ISDIR(mdd_object_type(src_obj)))
 643                 RETURN(-EPERM);
 644
 645         LASSERT(src_obj != tgt_obj);
 646         rc = mdd_may_create(env, tgt_obj, tattr, NULL, true);
 647         if (rc != 0)
 648                 RETURN(rc);
 649
 650         rc = __mdd_may_link(env, src_obj, cattr);
 651
 652         RETURN(rc);
 653 }
 654
 655 static int __mdd_index_delete_only(const struct lu_env *env, struct mdd_object *pobj,
 656                                    const char *name, struct thandle *handle)
 657 {
 658         struct dt_object *next = mdd_object_child(pobj);
 659         int rc;
 660         ENTRY;
 661
 662         if (dt_try_as_dir(env, next))
 663                 rc = dt_delete(env, next, (struct dt_key *)name, handle);
 664         else
 665                 rc = -ENOTDIR;
 666
 667         RETURN(rc);
 668 }
 669
 670 static int __mdd_index_insert_only(const struct lu_env *env,
 671                                    struct mdd_object *pobj,
 672                                    const struct lu_fid *lf, __u32 type,
 673                                    const char *name,
 674                                    struct thandle *handle)
 675 {
 676         struct dt_object *next = mdd_object_child(pobj);
 677         int               rc;
 678         ENTRY;
 679
 680         if (dt_try_as_dir(env, next)) {
 681                 struct dt_insert_rec    *rec = &mdd_env_info(env)->mti_dt_rec;
 682                 struct lu_ucred         *uc  = lu_ucred_check(env);
 683                 int                      ignore_quota;
 684
 685                 rec->rec_fid = lf;
 686                 rec->rec_type = type;
 687                 ignore_quota = uc ? uc->uc_cap & CFS_CAP_SYS_RESOURCE_MASK : 1;
 688                 rc = dt_insert(env, next, (const struct dt_rec *)rec,
 689                                (const struct dt_key *)name, handle,
 690                                ignore_quota);
 691         } else {
 692                 rc = -ENOTDIR;
 693         }
 694         RETURN(rc);
 695 }
 696
 697 /* insert named index, add reference if isdir */
 698 static int __mdd_index_insert(const struct lu_env *env, struct mdd_object *pobj,
 699                               const struct lu_fid *lf, __u32 type,
 700                               const char *name, struct thandle *handle)
 701 {
 702         int rc;
 703         ENTRY;
 704
 705         rc = __mdd_index_insert_only(env, pobj, lf, type, name, handle);
 706         if (rc == 0 && S_ISDIR(type)) {
 707                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
 708                 mdo_ref_add(env, pobj, handle);
 709                 mdd_write_unlock(env, pobj);
 710         }
 711
 712         RETURN(rc);
 713 }
 714
 715 /* delete named index, drop reference if isdir */
 716 static int __mdd_index_delete(const struct lu_env *env, struct mdd_object *pobj,
 717                               const char *name, int is_dir,
 718                               struct thandle *handle)
 719 {
 720         int               rc;
 721         ENTRY;
 722
 723         rc = __mdd_index_delete_only(env, pobj, name, handle);
 724         if (rc == 0 && is_dir) {
 725                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
 726                 mdo_ref_del(env, pobj, handle);
 727                 mdd_write_unlock(env, pobj);
 728         }
 729
 730         RETURN(rc);
 731 }
 732
 733 static int mdd_llog_record_calc_size(const struct lu_env *env,
 734                                      const struct lu_name *tname,
 735                                      const struct lu_name *sname)
 736 {
 737         const struct lu_ucred   *uc = lu_ucred(env);
 738         enum changelog_rec_flags crf = 0;
 739         size_t                   hdr_size = sizeof(struct llog_changelog_rec) -
 740                                             sizeof(struct changelog_rec);
 741
 742         if (sname != NULL)
 743                 crf |= CLF_RENAME;
 744
 745         if (uc != NULL && uc->uc_jobid[0] != '\0')
 746                 crf |= CLF_JOBID;
 747
 748         return llog_data_len(hdr_size + changelog_rec_offset(crf) +
 749                              (tname != NULL ? tname->ln_namelen : 0) +
 750                              (sname != NULL ? 1 + sname->ln_namelen : 0));
 751 }
 752
 753 int mdd_declare_changelog_store(const struct lu_env *env,
 754                                 struct mdd_device *mdd,
 755                                 const struct lu_name *tname,
 756                                 const struct lu_name *sname,
 757                                 struct thandle *handle)
 758 {
 759         struct obd_device               *obd = mdd2obd_dev(mdd);
 760         struct llog_ctxt                *ctxt;
 761         struct llog_changelog_rec       *rec;
 762         struct lu_buf                   *buf;
 763         struct thandle                  *llog_th;
 764         int                              reclen;
 765         int                              rc;
 766
 767         /* Not recording */
 768         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 769                 return 0;
 770
 771         reclen = mdd_llog_record_calc_size(env, tname, sname);
 772         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
 773         if (buf->lb_buf == NULL)
 774                 return -ENOMEM;
 775
 776         rec = buf->lb_buf;
 777         rec->cr_hdr.lrh_len = reclen;
 778         rec->cr_hdr.lrh_type = CHANGELOG_REC;
 779
 780         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
 781         if (ctxt == NULL)
 782                 return -ENXIO;
 783
 784         llog_th = thandle_get_sub(env, handle, ctxt->loc_handle->lgh_obj);
 785         if (IS_ERR(llog_th))
 786                 GOTO(out_put, rc = PTR_ERR(llog_th));
 787
 788         rc = llog_declare_add(env, ctxt->loc_handle, &rec->cr_hdr, llog_th);
 789
 790 out_put:
 791         llog_ctxt_put(ctxt);
 792
 793         return rc;
 794 }
 795
 796 /** Add a changelog entry \a rec to the changelog llog
 797  * \param mdd
 798  * \param rec
 799  * \param handle - currently ignored since llogs start their own transaction;
 800  *              this will hopefully be fixed in llog rewrite
 801  * \retval 0 ok
 802  */
 803 int mdd_changelog_store(const struct lu_env *env, struct mdd_device *mdd,
 804                         struct llog_changelog_rec *rec, struct thandle *th)
 805 {
 806         struct obd_device       *obd = mdd2obd_dev(mdd);
 807         struct llog_ctxt        *ctxt;
 808         struct thandle          *llog_th;
 809         int                      rc;
 810
 811         rec->cr_hdr.lrh_len = llog_data_len(sizeof(*rec) +
 812                                             changelog_rec_varsize(&rec->cr));
 813
 814         /* llog_lvfs_write_rec sets the llog tail len */
 815         rec->cr_hdr.lrh_type = CHANGELOG_REC;
 816         rec->cr.cr_time = cl_time();
 817
 818         spin_lock(&mdd->mdd_cl.mc_lock);
 819         /* NB: I suppose it's possible llog_add adds out of order wrt cr_index,
 820          * but as long as the MDD transactions are ordered correctly for e.g.
 821          * rename conflicts, I don't think this should matter. */
 822         rec->cr.cr_index = ++mdd->mdd_cl.mc_index;
 823         spin_unlock(&mdd->mdd_cl.mc_lock);
 824
 825         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
 826         if (ctxt == NULL)
 827                 return -ENXIO;
 828
 829         llog_th = thandle_get_sub(env, th, ctxt->loc_handle->lgh_obj);
 830         if (IS_ERR(llog_th))
 831                 GOTO(out_put, rc = PTR_ERR(llog_th));
 832
 833         /* nested journal transaction */
 834         rc = llog_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL, llog_th);
 835
 836 out_put:
 837         llog_ctxt_put(ctxt);
 838         if (rc > 0)
 839                 rc = 0;
 840         return rc;
 841 }
 842
 843 static void mdd_changelog_rec_ext_rename(struct changelog_rec *rec,
 844                                          const struct lu_fid *sfid,
 845                                          const struct lu_fid *spfid,
 846                                          const struct lu_name *sname)
 847 {
 848         struct changelog_ext_rename     *rnm = changelog_rec_rename(rec);
 849         size_t                           extsize = sname->ln_namelen + 1;
 850
 851         LASSERT(sfid != NULL);
 852         LASSERT(spfid != NULL);
 853         LASSERT(sname != NULL);
 854
 855         rnm->cr_sfid = *sfid;
 856         rnm->cr_spfid = *spfid;
 857
 858         changelog_rec_name(rec)[rec->cr_namelen] = '\0';
 859         strlcpy(changelog_rec_sname(rec), sname->ln_name, extsize);
 860         rec->cr_namelen += extsize;
 861 }
 862
 863 void mdd_changelog_rec_ext_jobid(struct changelog_rec *rec, const char *jobid)
 864 {
 865         struct changelog_ext_jobid      *jid = changelog_rec_jobid(rec);
 866
 867         if (jobid == NULL || jobid[0] == '\0')
 868                 return;
 869
 870         strlcpy(jid->cr_jobid, jobid, sizeof(jid->cr_jobid));
 871 }
 872
 873 /** Store a namespace change changelog record
 874  * If this fails, we must fail the whole transaction; we don't
 875  * want the change to commit without the log entry.
 876  * \param target - mdd_object of change
 877  * \param tpfid - target parent dir/object fid
 878  * \param sfid - source object fid
 879  * \param spfid - source parent fid
 880  * \param tname - target name string
 881  * \param sname - source name string
 882  * \param handle - transaction handle
 883  */
 884 int mdd_changelog_ns_store(const struct lu_env *env,
 885                            struct mdd_device *mdd,
 886                            enum changelog_rec_type type,
 887                            enum changelog_rec_flags crf,
 888                            struct mdd_object *target,
 889                            const struct lu_fid *tpfid,
 890                            const struct lu_fid *sfid,
 891                            const struct lu_fid *spfid,
 892                            const struct lu_name *tname,
 893                            const struct lu_name *sname,
 894                            struct thandle *handle)
 895 {
 896         const struct lu_ucred           *uc = lu_ucred(env);
 897         struct llog_changelog_rec       *rec;
 898         struct lu_buf                   *buf;
 899         int                              reclen;
 900         int                              rc;
 901         ENTRY;
 902
 903         /* Not recording */
 904         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 905                 RETURN(0);
 906
 907         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
 908                 RETURN(0);
 909
 910         LASSERT(tpfid != NULL);
 911         LASSERT(tname != NULL);
 912         LASSERT(handle != NULL);
 913
 914         reclen = mdd_llog_record_calc_size(env, tname, sname);
 915         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
 916         if (buf->lb_buf == NULL)
 917                 RETURN(-ENOMEM);
 918         rec = buf->lb_buf;
 919
 920         crf &= CLF_FLAGMASK;
 921
 922         if (uc != NULL && uc->uc_jobid[0] != '\0')
 923                 crf |= CLF_JOBID;
 924
 925         if (sname != NULL)
 926                 crf |= CLF_RENAME;
 927         else
 928                 crf |= CLF_VERSION;
 929
 930         rec->cr.cr_flags = crf;
 931         rec->cr.cr_type = (__u32)type;
 932         rec->cr.cr_pfid = *tpfid;
 933         rec->cr.cr_namelen = tname->ln_namelen;
 934         memcpy(changelog_rec_name(&rec->cr), tname->ln_name, tname->ln_namelen);
 935
 936         if (crf & CLF_RENAME)
 937                 mdd_changelog_rec_ext_rename(&rec->cr, sfid, spfid, sname);
 938
 939         if (crf & CLF_JOBID)
 940                 mdd_changelog_rec_ext_jobid(&rec->cr, uc->uc_jobid);
 941
 942         if (likely(target != NULL)) {
 943                 rec->cr.cr_tfid = *mdo2fid(target);
 944                 target->mod_cltime = cfs_time_current_64();
 945         } else {
 946                 fid_zero(&rec->cr.cr_tfid);
 947         }
 948
 949         rc = mdd_changelog_store(env, mdd, rec, handle);
 950         if (rc < 0) {
 951                 CERROR("%s: cannot store changelog record: type = %d, "
 952                        "name = '%s', t = "DFID", p = "DFID": rc = %d\n",
 953                        mdd2obd_dev(mdd)->obd_name, type, tname->ln_name,
 954                        PFID(&rec->cr.cr_tfid), PFID(&rec->cr.cr_pfid), rc);
 955                 return -EFAULT;
 956         }
 957
 958         return 0;
 959 }
 960
 961 static int __mdd_links_add(const struct lu_env *env,
 962                            struct mdd_object *mdd_obj,
 963                            struct linkea_data *ldata,
 964                            const struct lu_name *lname,
 965                            const struct lu_fid *pfid,
 966                            int first, int check)
 967 {
 968         int rc;
 969
 970         if (ldata->ld_leh == NULL) {
 971                 rc = first ? -ENODATA : mdd_links_read(env, mdd_obj, ldata);
 972                 if (rc) {
 973                         if (rc != -ENODATA)
 974                                 return rc;
 975                         rc = linkea_data_new(ldata,
 976                                              &mdd_env_info(env)->mti_link_buf);
 977                         if (rc)
 978                                 return rc;
 979                 }
 980         }
 981
 982         if (check) {
 983                 rc = linkea_links_find(ldata, lname, pfid);
 984                 if (rc && rc != -ENOENT)
 985                         return rc;
 986                 if (rc == 0)
 987                         return -EEXIST;
 988         }
 989
 990         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_MORE)) {
 991                 struct lu_fid *tfid = &mdd_env_info(env)->mti_fid2;
 992
 993                 *tfid = *pfid;
 994                 tfid->f_ver = ~0;
 995                 linkea_add_buf(ldata, lname, tfid);
 996         }
 997
 998         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LINKEA_MORE2))
 999                 linkea_add_buf(ldata, lname, pfid);
1000
1001         return linkea_add_buf(ldata, lname, pfid);
1002 }
1003
1004 static int __mdd_links_del(const struct lu_env *env,
1005                            struct mdd_object *mdd_obj,
1006                            struct linkea_data *ldata,
1007                            const struct lu_name *lname,
1008                            const struct lu_fid *pfid)
1009 {
1010         int rc;
1011
1012         if (ldata->ld_leh == NULL) {
1013                 rc = mdd_links_read(env, mdd_obj, ldata);
1014                 if (rc)
1015                         return rc;
1016         }
1017
1018         rc = linkea_links_find(ldata, lname, pfid);
1019         if (rc)
1020                 return rc;
1021
1022         linkea_del_buf(ldata, lname);
1023         return 0;
1024 }
1025
1026 static int mdd_linkea_prepare(const struct lu_env *env,
1027                               struct mdd_object *mdd_obj,
1028                               const struct lu_fid *oldpfid,
1029                               const struct lu_name *oldlname,
1030                               const struct lu_fid *newpfid,
1031                               const struct lu_name *newlname,
1032                               int first, int check,
1033                               struct linkea_data *ldata)
1034 {
1035         int rc = 0;
1036         ENTRY;
1037
1038         if (OBD_FAIL_CHECK(OBD_FAIL_FID_IGIF))
1039                 RETURN(0);
1040
1041         LASSERT(oldpfid != NULL || newpfid != NULL);
1042
1043         if (mdd_obj->mod_flags & DEAD_OBJ)
1044                 /* Unnecessary to update linkEA for dead object.  */
1045                 RETURN(0);
1046
1047         if (oldpfid != NULL) {
1048                 rc = __mdd_links_del(env, mdd_obj, ldata, oldlname, oldpfid);
1049                 if (rc) {
1050                         if ((check == 1) || (rc != -ENODATA && rc != -ENOENT))
1051                                 RETURN(rc);
1052
1053                         /* No changes done. */
1054                         rc = 0;
1055                 }
1056         }
1057
1058         /* If renaming, add the new record */
1059         if (newpfid != NULL)
1060                 rc = __mdd_links_add(env, mdd_obj, ldata, newlname, newpfid,
1061                                      first, check);
1062
1063         RETURN(rc);
1064 }
1065
1066 int mdd_links_rename(const struct lu_env *env,
1067                      struct mdd_object *mdd_obj,
1068                      const struct lu_fid *oldpfid,
1069                      const struct lu_name *oldlname,
1070                      const struct lu_fid *newpfid,
1071                      const struct lu_name *newlname,
1072                      struct thandle *handle,
1073                      struct linkea_data *ldata,
1074                      int first, int check)
1075 {
1076         int rc = 0;
1077         ENTRY;
1078
1079         if (ldata == NULL) {
1080                 ldata = &mdd_env_info(env)->mti_link_data;
1081                 memset(ldata, 0, sizeof(*ldata));
1082                 rc = mdd_linkea_prepare(env, mdd_obj, oldpfid, oldlname,
1083                                         newpfid, newlname, first, check, ldata);
1084                 if (rc)
1085                         GOTO(out, rc);
1086         }
1087
1088         if (!(mdd_obj->mod_flags & DEAD_OBJ))
1089                 rc = mdd_links_write(env, mdd_obj, ldata, handle);
1090
1091         GOTO(out, rc);
1092
1093 out:
1094         if (rc != 0) {
1095                 if (newlname == NULL)
1096                         CERROR("link_ea add failed %d "DFID"\n",
1097                                rc, PFID(mdd_object_fid(mdd_obj)));
1098                 else if (oldpfid == NULL)
1099                         CERROR("link_ea add '%.*s' failed %d "DFID"\n",
1100                                newlname->ln_namelen, newlname->ln_name, rc,
1101                                PFID(mdd_object_fid(mdd_obj)));
1102                 else if (newpfid == NULL)
1103                         CERROR("link_ea del '%.*s' failed %d "DFID"\n",
1104                                oldlname->ln_namelen, oldlname->ln_name, rc,
1105                                PFID(mdd_object_fid(mdd_obj)));
1106                 else
1107                         CERROR("link_ea rename '%.*s'->'%.*s' failed %d "DFID
1108                                "\n", oldlname->ln_namelen, oldlname->ln_name,
1109                                newlname->ln_namelen, newlname->ln_name, rc,
1110                                PFID(mdd_object_fid(mdd_obj)));
1111         }
1112
1113         if (is_vmalloc_addr(ldata->ld_buf))
1114                 /* if we vmalloced a large buffer drop it */
1115                 lu_buf_free(ldata->ld_buf);
1116
1117         return rc;
1118 }
1119
1120 static inline int mdd_links_add(const struct lu_env *env,
1121                                 struct mdd_object *mdd_obj,
1122                                 const struct lu_fid *pfid,
1123                                 const struct lu_name *lname,
1124                                 struct thandle *handle,
1125                                 struct linkea_data *ldata, int first)
1126 {
1127         return mdd_links_rename(env, mdd_obj, NULL, NULL,
1128                                 pfid, lname, handle, ldata, first, 0);
1129 }
1130
1131 static inline int mdd_links_del(const struct lu_env *env,
1132                                 struct mdd_object *mdd_obj,
1133                                 const struct lu_fid *pfid,
1134                                 const struct lu_name *lname,
1135                                 struct thandle *handle)
1136 {
1137         return mdd_links_rename(env, mdd_obj, pfid, lname,
1138                                 NULL, NULL, handle, NULL, 0, 0);
1139 }
1140
1141 /** Read the link EA into a temp buffer.
1142  * Uses the name_buf since it is generally large.
1143  * \retval IS_ERR err
1144  * \retval ptr to \a lu_buf (always \a mti_big_buf)
1145  */
1146 struct lu_buf *mdd_links_get(const struct lu_env *env,
1147                              struct mdd_object *mdd_obj)
1148 {
1149         struct linkea_data ldata = { NULL };
1150         int rc;
1151
1152         rc = mdd_links_read(env, mdd_obj, &ldata);
1153         return rc ? ERR_PTR(rc) : ldata.ld_buf;
1154 }
1155
1156 int mdd_links_write(const struct lu_env *env, struct mdd_object *mdd_obj,
1157                     struct linkea_data *ldata, struct thandle *handle)
1158 {
1159         const struct lu_buf *buf;
1160         int                 rc;
1161
1162         if (ldata == NULL || ldata->ld_buf == NULL ||
1163             ldata->ld_leh == NULL)
1164                 return 0;
1165
1166         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_LINKEA))
1167                 return 0;
1168
1169 again:
1170         buf = mdd_buf_get_const(env, ldata->ld_buf->lb_buf,
1171                                 ldata->ld_leh->leh_len);
1172         rc = mdo_xattr_set(env, mdd_obj, buf, XATTR_NAME_LINK, 0, handle);
1173         if (unlikely(rc == -ENOSPC)) {
1174                 rc = linkea_overflow_shrink(ldata);
1175                 if (likely(rc > 0))
1176                         goto again;
1177         }
1178
1179         return rc;
1180 }
1181
1182 static int mdd_declare_links_add(const struct lu_env *env,
1183                                  struct mdd_object *mdd_obj,
1184                                  struct thandle *handle,
1185                                  struct linkea_data *ldata)
1186 {
1187         int     rc;
1188         int     ea_len;
1189         void    *linkea;
1190
1191         if (ldata != NULL && ldata->ld_leh != NULL) {
1192                 ea_len = ldata->ld_leh->leh_len;
1193                 linkea = ldata->ld_buf->lb_buf;
1194         } else {
1195                 ea_len = MAX_LINKEA_SIZE;
1196                 linkea = NULL;
1197         }
1198
1199         rc = mdo_declare_xattr_set(env, mdd_obj,
1200                                    mdd_buf_get_const(env, linkea, ea_len),
1201                                    XATTR_NAME_LINK, 0, handle);
1202
1203         return rc;
1204 }
1205
1206 static inline int mdd_declare_links_del(const struct lu_env *env,
1207                                         struct mdd_object *c,
1208                                         struct thandle *handle)
1209 {
1210         int rc = 0;
1211
1212         /* For directory, the linkEA will be removed together
1213          * with the object. */
1214         if (!S_ISDIR(mdd_object_type(c)))
1215                 rc = mdd_declare_links_add(env, c, handle, NULL);
1216
1217         return rc;
1218 }
1219
1220 static int mdd_declare_link(const struct lu_env *env,
1221                             struct mdd_device *mdd,
1222                             struct mdd_object *p,
1223                             struct mdd_object *c,
1224                             const struct lu_name *name,
1225                             struct thandle *handle,
1226                             struct lu_attr *la,
1227                             struct linkea_data *data)
1228 {
1229         struct lu_fid tfid = *mdo2fid(c);
1230         int rc;
1231
1232         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING3))
1233                 tfid.f_oid = cfs_fail_val;
1234
1235         rc = mdo_declare_index_insert(env, p, &tfid, mdd_object_type(c),
1236                                       name->ln_name, handle);
1237         if (rc != 0)
1238                 return rc;
1239
1240         rc = mdo_declare_ref_add(env, c, handle);
1241         if (rc != 0)
1242                 return rc;
1243
1244         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MORE_NLINK)) {
1245                 rc = mdo_declare_ref_add(env, c, handle);
1246                 if (rc != 0)
1247                         return rc;
1248         }
1249
1250         la->la_valid = LA_CTIME | LA_MTIME;
1251         rc = mdo_declare_attr_set(env, p, la, handle);
1252         if (rc != 0)
1253                 return rc;
1254
1255         la->la_valid = LA_CTIME;
1256         rc = mdo_declare_attr_set(env, c, la, handle);
1257         if (rc != 0)
1258                 return rc;
1259
1260         rc = mdd_declare_links_add(env, c, handle, data);
1261         if (rc != 0)
1262                 return rc;
1263
1264         rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
1265
1266         return rc;
1267 }
1268
1269 static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
1270                     struct md_object *src_obj, const struct lu_name *lname,
1271                     struct md_attr *ma)
1272 {
1273         const char *name = lname->ln_name;
1274         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1275         struct mdd_object *mdd_tobj = md2mdd_obj(tgt_obj);
1276         struct mdd_object *mdd_sobj = md2mdd_obj(src_obj);
1277         struct lu_attr    *cattr = MDD_ENV_VAR(env, cattr);
1278         struct lu_attr    *tattr = MDD_ENV_VAR(env, tattr);
1279         struct mdd_device *mdd = mdo2mdd(src_obj);
1280         struct thandle *handle;
1281         struct lu_fid *tfid = &mdd_env_info(env)->mti_fid2;
1282         struct linkea_data *ldata = &mdd_env_info(env)->mti_link_data;
1283         int rc;
1284         ENTRY;
1285
1286         rc = mdd_la_get(env, mdd_sobj, cattr);
1287         if (rc != 0)
1288                 RETURN(rc);
1289
1290         rc = mdd_la_get(env, mdd_tobj, tattr);
1291         if (rc != 0)
1292                 RETURN(rc);
1293
1294         handle = mdd_trans_create(env, mdd);
1295         if (IS_ERR(handle))
1296                 GOTO(out_pending, rc = PTR_ERR(handle));
1297
1298         memset(ldata, 0, sizeof(*ldata));
1299
1300         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1301         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1302
1303         /* Note: even this function will change ldata, but it comes from
1304          * thread_info, which is completely temporary and only seen in
1305          * this function, so we do not need reset ldata once it fails.*/
1306         rc = mdd_linkea_prepare(env, mdd_sobj, NULL, NULL, mdo2fid(mdd_tobj),
1307                                 lname, 0, 0, ldata);
1308         if (rc != 0)
1309                 GOTO(stop, rc);
1310
1311         rc = mdd_declare_link(env, mdd, mdd_tobj, mdd_sobj, lname, handle,
1312                               la, ldata);
1313         if (rc)
1314                 GOTO(stop, rc);
1315
1316         rc = mdd_trans_start(env, mdd, handle);
1317         if (rc)
1318                 GOTO(stop, rc);
1319
1320         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
1321         rc = mdd_link_sanity_check(env, mdd_tobj, tattr, lname, mdd_sobj,
1322                                    cattr);
1323         if (rc)
1324                 GOTO(out_unlock, rc);
1325
1326         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_LESS_NLINK)) {
1327                 rc = mdo_ref_add(env, mdd_sobj, handle);
1328                 if (rc != 0)
1329                         GOTO(out_unlock, rc);
1330         }
1331
1332         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MORE_NLINK)) {
1333                 rc = mdo_ref_add(env, mdd_sobj, handle);
1334                 if (rc != 0)
1335                         GOTO(out_unlock, rc);
1336         }
1337
1338         *tfid = *mdo2fid(mdd_sobj);
1339         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING3))
1340                 tfid->f_oid = cfs_fail_val;
1341
1342         rc = __mdd_index_insert_only(env, mdd_tobj, tfid,
1343                                      mdd_object_type(mdd_sobj), name, handle);
1344         if (rc != 0) {
1345                 mdo_ref_del(env, mdd_sobj, handle);
1346                 GOTO(out_unlock, rc);
1347         }
1348
1349         la->la_valid = LA_CTIME | LA_MTIME;
1350         rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
1351         if (rc)
1352                 GOTO(out_unlock, rc);
1353
1354         la->la_valid = LA_CTIME;
1355         rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
1356         if (rc == 0)
1357                 /* Note: The failure of links_add should not cause the
1358                  * link failure, so do not check return value. */
1359                 mdd_links_add(env, mdd_sobj, mdo2fid(mdd_tobj),
1360                               lname, handle, ldata, 0);
1361
1362         EXIT;
1363 out_unlock:
1364         mdd_write_unlock(env, mdd_sobj);
1365         if (rc == 0)
1366                 rc = mdd_changelog_ns_store(env, mdd, CL_HARDLINK, 0, mdd_sobj,
1367                                             mdo2fid(mdd_tobj), NULL, NULL,
1368                                             lname, NULL, handle);
1369 stop:
1370         rc = mdd_trans_stop(env, mdd, rc, handle);
1371         if (is_vmalloc_addr(ldata->ld_buf))
1372                 /* if we vmalloced a large buffer drop it */
1373                 lu_buf_free(ldata->ld_buf);
1374 out_pending:
1375         return rc;
1376 }
1377
1378 static int mdd_mark_orphan_object(const struct lu_env *env,
1379                                 struct mdd_object *obj, struct thandle *handle,
1380                                 bool declare)
1381 {
1382         struct lu_attr *attr = MDD_ENV_VAR(env, la_for_start);
1383         int rc;
1384
1385         if (!S_ISDIR(mdd_object_type(obj)))
1386                 return 0;
1387
1388         attr->la_valid = LA_FLAGS;
1389         attr->la_flags = LUSTRE_ORPHAN_FL;
1390
1391         if (declare)
1392                 rc = mdo_declare_attr_set(env, obj, attr, handle);
1393         else
1394                 rc = mdo_attr_set(env, obj, attr, handle);
1395
1396         return rc;
1397 }
1398
1399 static int mdd_declare_finish_unlink(const struct lu_env *env,
1400                                      struct mdd_object *obj,
1401                                      struct thandle *handle)
1402 {
1403         int     rc;
1404
1405         /* Sigh, we do not know if the unlink object will become orphan in
1406          * declare phase, but fortunately the flags here does not matter
1407          * in current declare implementation */
1408         rc = mdd_mark_orphan_object(env, obj, handle, true);
1409         if (rc != 0)
1410                 return rc;
1411
1412         rc = mdo_declare_destroy(env, obj, handle);
1413         if (rc != 0)
1414                 return rc;
1415
1416         rc = orph_declare_index_insert(env, obj, mdd_object_type(obj), handle);
1417         if (rc != 0)
1418                 return rc;
1419
1420         return mdd_declare_links_del(env, obj, handle);
1421 }
1422
1423 /* caller should take a lock before calling */
1424 int mdd_finish_unlink(const struct lu_env *env,
1425                       struct mdd_object *obj, struct md_attr *ma,
1426                       const struct mdd_object *pobj,
1427                       const struct lu_name *lname,
1428                       struct thandle *th)
1429 {
1430         int rc = 0;
1431         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1432         ENTRY;
1433
1434         LASSERT(mdd_write_locked(env, obj) != 0);
1435
1436         if (ma->ma_attr.la_nlink == 0 || is_dir) {
1437                 /* add new orphan and the object
1438                  * will be deleted during mdd_close() */
1439                 obj->mod_flags |= DEAD_OBJ;
1440                 if (obj->mod_count) {
1441                         rc = __mdd_orphan_add(env, obj, th);
1442                         if (rc == 0)
1443                                 CDEBUG(D_HA, "Object "DFID" is inserted into "
1444                                         "orphan list, open count = %d\n",
1445                                         PFID(mdd_object_fid(obj)),
1446                                         obj->mod_count);
1447                         else
1448                                 CERROR("Object "DFID" fail to be an orphan, "
1449                                        "open count = %d, maybe cause failed "
1450                                        "open replay\n",
1451                                         PFID(mdd_object_fid(obj)),
1452                                         obj->mod_count);
1453
1454                         /* mark object as an orphan here, not
1455                          * before __mdd_orphan_add() as racing
1456                          * mdd_la_get() may propagate ORPHAN_OBJ
1457                          * causing the asserition */
1458                         rc = mdd_mark_orphan_object(env, obj, th, false);
1459                 } else {
1460                         rc = mdo_destroy(env, obj, th);
1461                 }
1462         } else if (!is_dir) {
1463                 /* old files may not have link ea; ignore errors */
1464                 mdd_links_del(env, obj, mdo2fid(pobj), lname, th);
1465         }
1466
1467         RETURN(rc);
1468 }
1469
1470 /*
1471  * pobj maybe NULL
1472  * has mdd_write_lock on cobj already, but not on pobj yet
1473  */
1474 int mdd_unlink_sanity_check(const struct lu_env *env, struct mdd_object *pobj,
1475                             const struct lu_attr *pattr,
1476                             struct mdd_object *cobj,
1477                             const struct lu_attr *cattr)
1478 {
1479         int rc;
1480         ENTRY;
1481
1482         rc = mdd_may_delete(env, pobj, pattr, cobj, cattr, NULL, 1, 1);
1483
1484         RETURN(rc);
1485 }
1486
1487 static int mdd_declare_unlink(const struct lu_env *env, struct mdd_device *mdd,
1488                               struct mdd_object *p, struct mdd_object *c,
1489                               const struct lu_name *name, struct md_attr *ma,
1490                               struct thandle *handle, int no_name, int is_dir)
1491 {
1492         struct lu_attr  *la = &mdd_env_info(env)->mti_la_for_fix;
1493         int              rc;
1494
1495         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1496                 if (likely(no_name == 0)) {
1497                         rc = mdo_declare_index_delete(env, p, name->ln_name,
1498                                                       handle);
1499                         if (rc != 0)
1500                                 return rc;
1501                 }
1502
1503                 if (is_dir != 0) {
1504                         rc = mdo_declare_ref_del(env, p, handle);
1505                         if (rc != 0)
1506                                 return rc;
1507                 }
1508         }
1509
1510         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1511         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1512         la->la_valid = LA_CTIME | LA_MTIME;
1513         rc = mdo_declare_attr_set(env, p, la, handle);
1514         if (rc)
1515                 return rc;
1516
1517         if (c != NULL) {
1518                 rc = mdo_declare_ref_del(env, c, handle);
1519                 if (rc)
1520                         return rc;
1521
1522                 rc = mdo_declare_ref_del(env, c, handle);
1523                 if (rc)
1524                         return rc;
1525
1526                 la->la_valid = LA_CTIME;
1527                 rc = mdo_declare_attr_set(env, c, la, handle);
1528                 if (rc)
1529                         return rc;
1530
1531                 rc = mdd_declare_finish_unlink(env, c, handle);
1532                 if (rc)
1533                         return rc;
1534
1535                 /* FIXME: need changelog for remove entry */
1536                 rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
1537         }
1538
1539         return rc;
1540 }
1541
1542 /*
1543  * test if a file has an HSM archive
1544  * if HSM attributes are not found in ma update them from
1545  * HSM xattr
1546  */
1547 static bool mdd_hsm_archive_exists(const struct lu_env *env,
1548                                    struct mdd_object *obj,
1549                                    struct md_attr *ma)
1550 {
1551         ENTRY;
1552
1553         if (!(ma->ma_valid & MA_HSM)) {
1554                 /* no HSM MD provided, read xattr */
1555                 struct lu_buf   *hsm_buf;
1556                 const size_t     buflen = sizeof(struct hsm_attrs);
1557                 int              rc;
1558
1559                 hsm_buf = mdd_buf_get(env, NULL, 0);
1560                 lu_buf_alloc(hsm_buf, buflen);
1561                 rc = mdo_xattr_get(env, obj, hsm_buf, XATTR_NAME_HSM);
1562                 rc = lustre_buf2hsm(hsm_buf->lb_buf, rc, &ma->ma_hsm);
1563                 lu_buf_free(hsm_buf);
1564                 if (rc < 0)
1565                         RETURN(false);
1566
1567                 ma->ma_valid |= MA_HSM;
1568         }
1569         if (ma->ma_hsm.mh_flags & HS_EXISTS)
1570                 RETURN(true);
1571         RETURN(false);
1572 }
1573
1574 /**
1575  * Delete name entry and the object.
1576  * Note: no_name == 1 means it only destory the object, i.e. name_entry
1577  * does not exist for this object, and it could only happen during resending
1578  * of remote unlink. see the comments in mdt_reint_unlink. Unfortunately, lname
1579  * is also needed in this case(needed by changelog), so we have to add another
1580  * parameter(no_name)here. XXX: this is only needed in DNE phase I, on Phase II,
1581  * the ENOENT failure should be able to be fixed by redo mechanism.
1582  */
1583 static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
1584                       struct md_object *cobj, const struct lu_name *lname,
1585                       struct md_attr *ma, int no_name)
1586 {
1587         const char *name = lname->ln_name;
1588         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
1589         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
1590         struct lu_attr *la = &mdd_env_info(env)->mti_la_for_fix;
1591         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1592         struct mdd_object *mdd_cobj = NULL;
1593         struct mdd_device *mdd = mdo2mdd(pobj);
1594         struct thandle    *handle;
1595         int rc, is_dir = 0, cl_flags = 0;
1596         ENTRY;
1597
1598         /* cobj == NULL means only delete name entry */
1599         if (likely(cobj != NULL)) {
1600                 mdd_cobj = md2mdd_obj(cobj);
1601                 if (mdd_object_exists(mdd_cobj) == 0)
1602                         RETURN(-ENOENT);
1603         }
1604
1605         rc = mdd_la_get(env, mdd_pobj, pattr);
1606         if (rc)
1607                 RETURN(rc);
1608
1609         if (likely(mdd_cobj != NULL)) {
1610                 /* fetch cattr */
1611                 rc = mdd_la_get(env, mdd_cobj, cattr);
1612                 if (rc)
1613                         RETURN(rc);
1614
1615                 is_dir = S_ISDIR(cattr->la_mode);
1616                 /* search for an existing archive.
1617                  * we should check ahead as the object
1618                  * can be destroyed in this transaction */
1619                 if (mdd_hsm_archive_exists(env, mdd_cobj, ma))
1620                         cl_flags |= CLF_UNLINK_HSM_EXISTS;
1621         }
1622
1623         rc = mdd_unlink_sanity_check(env, mdd_pobj, pattr, mdd_cobj, cattr);
1624         if (rc)
1625                 RETURN(rc);
1626
1627         handle = mdd_trans_create(env, mdd);
1628         if (IS_ERR(handle))
1629                 RETURN(PTR_ERR(handle));
1630
1631         rc = mdd_declare_unlink(env, mdd, mdd_pobj, mdd_cobj,
1632                                 lname, ma, handle, no_name, is_dir);
1633         if (rc)
1634                 GOTO(stop, rc);
1635
1636         rc = mdd_trans_start(env, mdd, handle);
1637         if (rc)
1638                 GOTO(stop, rc);
1639
1640         if (likely(mdd_cobj != NULL))
1641                 mdd_write_lock(env, mdd_cobj, MOR_TGT_CHILD);
1642
1643         if (likely(no_name == 0) && !OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1644                 rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle);
1645                 if (rc)
1646                         GOTO(cleanup, rc);
1647         }
1648
1649         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MUL_REF) ||
1650             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_NAMEENTRY))
1651                 GOTO(cleanup, rc = 0);
1652
1653         if (likely(mdd_cobj != NULL)) {
1654                 rc = mdo_ref_del(env, mdd_cobj, handle);
1655                 if (rc != 0) {
1656                         __mdd_index_insert_only(env, mdd_pobj,
1657                                                 mdo2fid(mdd_cobj),
1658                                                 mdd_object_type(mdd_cobj),
1659                                                 name, handle);
1660                         GOTO(cleanup, rc);
1661                 }
1662
1663                 if (is_dir)
1664                         /* unlink dot */
1665                         mdo_ref_del(env, mdd_cobj, handle);
1666
1667                 /* fetch updated nlink */
1668                 rc = mdd_la_get(env, mdd_cobj, cattr);
1669                 if (rc)
1670                         GOTO(cleanup, rc);
1671         }
1672
1673         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1674         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1675
1676         la->la_valid = LA_CTIME | LA_MTIME;
1677         rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
1678         if (rc)
1679                 GOTO(cleanup, rc);
1680
1681         /* Enough for only unlink the entry */
1682         if (unlikely(mdd_cobj == NULL))
1683                 GOTO(stop, rc);
1684
1685         if (cattr->la_nlink > 0 || mdd_cobj->mod_count > 0) {
1686                 /* update ctime of an unlinked file only if it is still
1687                  * opened or a link still exists */
1688                 la->la_valid = LA_CTIME;
1689                 rc = mdd_update_time(env, mdd_cobj, cattr, la, handle);
1690                 if (rc)
1691                         GOTO(cleanup, rc);
1692         }
1693
1694         /* XXX: this transfer to ma will be removed with LOD/OSP */
1695         ma->ma_attr = *cattr;
1696         ma->ma_valid |= MA_INODE;
1697         rc = mdd_finish_unlink(env, mdd_cobj, ma, mdd_pobj, lname, handle);
1698         if (rc != 0)
1699                 GOTO(cleanup, rc);
1700
1701         /* fetch updated nlink */
1702         rc = mdd_la_get(env, mdd_cobj, cattr);
1703         /* if object is removed then we can't get its attrs,
1704          * use last get */
1705         if (rc == -ENOENT) {
1706                 cattr->la_nlink = 0;
1707                 rc = 0;
1708         }
1709
1710         if (cattr->la_nlink == 0) {
1711                 ma->ma_attr = *cattr;
1712                 ma->ma_valid |= MA_INODE;
1713         }
1714
1715         EXIT;
1716 cleanup:
1717         if (likely(mdd_cobj != NULL))
1718                 mdd_write_unlock(env, mdd_cobj);
1719
1720         if (rc == 0) {
1721                 if (cattr->la_nlink == 0)
1722                         cl_flags |= CLF_UNLINK_LAST;
1723                 else
1724                         cl_flags &= ~CLF_UNLINK_HSM_EXISTS;
1725
1726                 rc = mdd_changelog_ns_store(env, mdd,
1727                         is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,
1728                         mdd_cobj, mdo2fid(mdd_pobj), NULL, NULL, lname, NULL,
1729                         handle);
1730         }
1731
1732 stop:
1733         rc = mdd_trans_stop(env, mdd, rc, handle);
1734
1735         return rc;
1736 }
1737
1738 /*
1739  * The permission has been checked when obj created, no need check again.
1740  */
1741 static int mdd_cd_sanity_check(const struct lu_env *env,
1742                                struct mdd_object *obj)
1743 {
1744         ENTRY;
1745
1746         /* EEXIST check */
1747         if (!obj || mdd_is_dead_obj(obj))
1748                 RETURN(-ENOENT);
1749
1750         RETURN(0);
1751 }
1752
1753 static int mdd_create_data(const struct lu_env *env,
1754                            struct md_object *pobj,
1755                            struct md_object *cobj,
1756                            const struct md_op_spec *spec,
1757                            struct md_attr *ma)
1758 {
1759         struct mdd_device *mdd = mdo2mdd(cobj);
1760         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1761         struct mdd_object *son = md2mdd_obj(cobj);
1762         struct thandle    *handle;
1763         const struct lu_buf *buf;
1764         struct lu_attr    *attr = MDD_ENV_VAR(env, cattr);
1765         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
1766         int                rc;
1767         ENTRY;
1768
1769         rc = mdd_cd_sanity_check(env, son);
1770         if (rc)
1771                 RETURN(rc);
1772
1773         if (!md_should_create(spec->sp_cr_flags))
1774                 RETURN(0);
1775
1776         /*
1777          * there are following use cases for this function:
1778          * 1) late striping - file was created with MDS_OPEN_DELAY_CREATE
1779          *    striping can be specified or not
1780          * 2) CMD?
1781          */
1782         rc = mdd_la_get(env, son, attr);
1783         if (rc)
1784                 RETURN(rc);
1785
1786         /* calling ->ah_make_hint() is used to transfer information from parent */
1787         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
1788
1789         handle = mdd_trans_create(env, mdd);
1790         if (IS_ERR(handle))
1791                 GOTO(out_free, rc = PTR_ERR(handle));
1792
1793         /*
1794          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1795          * Should this be fixed?
1796          */
1797         CDEBUG(D_OTHER, "ea %p/%u, cr_flags %#llo, no_create %u\n",
1798                spec->u.sp_ea.eadata, spec->u.sp_ea.eadatalen,
1799                spec->sp_cr_flags, spec->no_create);
1800
1801         if (spec->no_create || (spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
1802                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
1803                                         spec->u.sp_ea.eadatalen);
1804         } else {
1805                 buf = &LU_BUF_NULL;
1806         }
1807
1808         rc = dt_declare_xattr_set(env, mdd_object_child(son), buf,
1809                                   XATTR_NAME_LOV, 0, handle);
1810         if (rc)
1811                 GOTO(stop, rc);
1812
1813         rc = mdd_declare_changelog_store(env, mdd, NULL, NULL, handle);
1814         if (rc)
1815                 GOTO(stop, rc);
1816
1817         rc = mdd_trans_start(env, mdd, handle);
1818         if (rc)
1819                 GOTO(stop, rc);
1820
1821         rc = dt_xattr_set(env, mdd_object_child(son), buf, XATTR_NAME_LOV,
1822                           0, handle);
1823
1824         if (rc)
1825                 GOTO(stop, rc);
1826
1827         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, son, handle);
1828
1829 stop:
1830         rc = mdd_trans_stop(env, mdd, rc, handle);
1831
1832 out_free:
1833         RETURN(rc);
1834 }
1835
1836 static int mdd_declare_object_initialize(const struct lu_env *env,
1837                                          struct mdd_object *parent,
1838                                          struct mdd_object *child,
1839                                          const struct lu_attr *attr,
1840                                          struct thandle *handle)
1841 {
1842         int rc;
1843         ENTRY;
1844
1845         LASSERT(attr->la_valid & (LA_MODE | LA_TYPE));
1846         if (!S_ISDIR(attr->la_mode))
1847                 RETURN(0);
1848
1849         rc = mdo_declare_index_insert(env, child, mdo2fid(child), S_IFDIR,
1850                                       dot, handle);
1851         if (rc != 0)
1852                 RETURN(rc);
1853
1854         rc = mdo_declare_ref_add(env, child, handle);
1855         if (rc != 0)
1856                 RETURN(rc);
1857
1858         rc = mdo_declare_index_insert(env, child, mdo2fid(parent), S_IFDIR,
1859                                       dotdot, handle);
1860
1861         RETURN(rc);
1862 }
1863
1864 static int mdd_object_initialize(const struct lu_env *env,
1865                                  const struct lu_fid *pfid,
1866                                  struct mdd_object *child,
1867                                  struct lu_attr *attr, struct thandle *handle,
1868                                  const struct md_op_spec *spec)
1869 {
1870         int rc = 0;
1871         ENTRY;
1872
1873         if (S_ISDIR(attr->la_mode)) {
1874                 /* Add "." and ".." for newly created dir */
1875                 mdo_ref_add(env, child, handle);
1876                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1877                                              S_IFDIR, dot, handle);
1878                 if (rc == 0)
1879                         rc = __mdd_index_insert_only(env, child, pfid, S_IFDIR,
1880                                                      dotdot, handle);
1881                 if (rc != 0)
1882                         mdo_ref_del(env, child, handle);
1883         }
1884
1885         RETURN(rc);
1886 }
1887
1888 /**
1889  * This function checks whether it can create a file/dir under the
1890  * directory(@pobj). The directory(@pobj) is not being locked by
1891  * mdd lock.
1892  *
1893  * \param[in] env       execution environment
1894  * \param[in] pobj      the directory to create files
1895  * \param[in] pattr     the attributes of the directory
1896  * \param[in] lname     the name of the created file/dir
1897  * \param[in] cattr     the attributes of the file/dir
1898  * \param[in] spec      create specification
1899  *
1900  * \retval              = 0 it is allowed to create file/dir under
1901  *                      the directory
1902  * \retval              negative error not allowed to create file/dir
1903  *                      under the directory
1904  */
1905 static int mdd_create_sanity_check(const struct lu_env *env,
1906                                    struct md_object *pobj,
1907                                    const struct lu_attr *pattr,
1908                                    const struct lu_name *lname,
1909                                    struct lu_attr *cattr,
1910                                    struct md_op_spec *spec)
1911 {
1912         struct mdd_thread_info *info = mdd_env_info(env);
1913         struct lu_fid     *fid       = &info->mti_fid;
1914         struct mdd_object *obj       = md2mdd_obj(pobj);
1915         struct mdd_device *m         = mdo2mdd(pobj);
1916         bool            check_perm = true;
1917         int rc;
1918         ENTRY;
1919
1920         /* EEXIST check */
1921         if (mdd_is_dead_obj(obj))
1922                 RETURN(-ENOENT);
1923
1924         /*
1925          * In some cases this lookup is not needed - we know before if name
1926          * exists or not because MDT performs lookup for it.
1927          * name length check is done in lookup.
1928          */
1929         if (spec->sp_cr_lookup) {
1930                 /*
1931                  * Check if the name already exist, though it will be checked in
1932                  * _index_insert also, for avoiding rolling back if exists
1933                  * _index_insert.
1934                  */
1935                 rc = __mdd_lookup(env, pobj, pattr, lname, fid,
1936                                   MAY_WRITE | MAY_EXEC);
1937                 if (rc != -ENOENT)
1938                         RETURN(rc ? : -EEXIST);
1939
1940                 /* Permission is already being checked in mdd_lookup */
1941                 check_perm = false;
1942         }
1943
1944         rc = mdd_may_create(env, obj, pattr, NULL, check_perm);
1945         if (rc != 0)
1946                 RETURN(rc);
1947
1948         /* sgid check */
1949         if (pattr->la_mode & S_ISGID) {
1950                 cattr->la_gid = pattr->la_gid;
1951                 if (S_ISDIR(cattr->la_mode)) {
1952                         cattr->la_mode |= S_ISGID;
1953                         cattr->la_valid |= LA_MODE;
1954                 }
1955         }
1956
1957         rc = mdd_name_check(m, lname);
1958         if (rc < 0)
1959                 RETURN(rc);
1960
1961         switch (cattr->la_mode & S_IFMT) {
1962         case S_IFLNK: {
1963                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1964
1965                 if (symlen > m->mdd_dt_conf.ddp_symlink_max)
1966                         RETURN(-ENAMETOOLONG);
1967                 else
1968                         RETURN(0);
1969         }
1970         case S_IFDIR:
1971         case S_IFREG:
1972         case S_IFCHR:
1973         case S_IFBLK:
1974         case S_IFIFO:
1975         case S_IFSOCK:
1976                 rc = 0;
1977                 break;
1978         default:
1979                 rc = -EINVAL;
1980                 break;
1981         }
1982         RETURN(rc);
1983 }
1984
1985 static int mdd_declare_object_create(const struct lu_env *env,
1986                                      struct mdd_device *mdd,
1987                                      struct mdd_object *p, struct mdd_object *c,
1988                                      struct lu_attr *attr,
1989                                      struct thandle *handle,
1990                                      const struct md_op_spec *spec,
1991                                      struct lu_buf *def_acl_buf,
1992                                      struct lu_buf *acl_buf,
1993                                      struct dt_allocation_hint *hint)
1994 {
1995         const struct lu_buf *buf;
1996         int rc;
1997
1998         rc = mdd_declare_object_create_internal(env, p, c, attr, handle, spec,
1999                                                 hint);
2000         if (rc)
2001                 GOTO(out, rc);
2002
2003 #ifdef CONFIG_FS_POSIX_ACL
2004         if (def_acl_buf->lb_len > 0 && S_ISDIR(attr->la_mode)) {
2005                 /* if dir, then can inherit default ACl */
2006                 rc = mdo_declare_xattr_set(env, c, def_acl_buf,
2007                                            XATTR_NAME_ACL_DEFAULT,
2008                                            0, handle);
2009                 if (rc)
2010                         GOTO(out, rc);
2011         }
2012
2013         if (acl_buf->lb_len > 0) {
2014                 rc = mdo_declare_attr_set(env, c, attr, handle);
2015                 if (rc)
2016                         GOTO(out, rc);
2017
2018                 rc = mdo_declare_xattr_set(env, c, acl_buf,
2019                                            XATTR_NAME_ACL_ACCESS, 0, handle);
2020                 if (rc)
2021                         GOTO(out, rc);
2022         }
2023 #endif
2024         rc = mdd_declare_object_initialize(env, p, c, attr, handle);
2025         if (rc)
2026                 GOTO(out, rc);
2027
2028         /* replay case, create LOV EA from client data */
2029         if (spec->no_create ||
2030             (spec->sp_cr_flags & MDS_OPEN_HAS_EA && S_ISREG(attr->la_mode))) {
2031                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2032                                         spec->u.sp_ea.eadatalen);
2033                 rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV, 0,
2034                                            handle);
2035                 if (rc)
2036                         GOTO(out, rc);
2037         }
2038
2039         if (S_ISLNK(attr->la_mode)) {
2040                 const char *target_name = spec->u.sp_symname;
2041                 int sym_len = strlen(target_name);
2042                 const struct lu_buf *buf;
2043
2044                 buf = mdd_buf_get_const(env, target_name, sym_len);
2045                 rc = dt_declare_record_write(env, mdd_object_child(c),
2046                                              buf, 0, handle);
2047                 if (rc)
2048                         GOTO(out, rc);
2049         }
2050
2051         if (spec->sp_cr_file_secctx_name != NULL) {
2052                 buf = mdd_buf_get_const(env, spec->sp_cr_file_secctx,
2053                                         spec->sp_cr_file_secctx_size);
2054                 rc = mdo_declare_xattr_set(env, c, buf,
2055                                            spec->sp_cr_file_secctx_name, 0,
2056                                            handle);
2057                 if (rc < 0)
2058                         GOTO(out, rc);
2059         }
2060 out:
2061         return rc;
2062 }
2063
2064 static int mdd_declare_create(const struct lu_env *env, struct mdd_device *mdd,
2065                               struct mdd_object *p, struct mdd_object *c,
2066                               const struct lu_name *name,
2067                               struct lu_attr *attr,
2068                               struct thandle *handle,
2069                               const struct md_op_spec *spec,
2070                               struct linkea_data *ldata,
2071                               struct lu_buf *def_acl_buf,
2072                               struct lu_buf *acl_buf,
2073                               struct dt_allocation_hint *hint)
2074 {
2075         int rc;
2076
2077         rc = mdd_declare_object_create(env, mdd, p, c, attr, handle, spec,
2078                                        def_acl_buf, acl_buf, hint);
2079         if (rc)
2080                 GOTO(out, rc);
2081
2082         if (S_ISDIR(attr->la_mode)) {
2083                 rc = mdo_declare_ref_add(env, p, handle);
2084                 if (rc)
2085                         GOTO(out, rc);
2086         }
2087
2088         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2089                 rc = orph_declare_index_insert(env, c, attr->la_mode, handle);
2090                 if (rc)
2091                         GOTO(out, rc);
2092         } else {
2093                 struct lu_attr  *la = &mdd_env_info(env)->mti_la_for_fix;
2094
2095                 rc = mdo_declare_index_insert(env, p, mdo2fid(c), attr->la_mode,
2096                                               name->ln_name, handle);
2097                 if (rc != 0)
2098                         return rc;
2099
2100                 rc = mdd_declare_links_add(env, c, handle, ldata);
2101                 if (rc)
2102                         return rc;
2103
2104                 *la = *attr;
2105                 la->la_valid = LA_CTIME | LA_MTIME;
2106                 rc = mdo_declare_attr_set(env, p, la, handle);
2107                 if (rc)
2108                         return rc;
2109
2110                 rc = mdd_declare_changelog_store(env, mdd, name, NULL, handle);
2111                 if (rc)
2112                         return rc;
2113         }
2114 out:
2115         return rc;
2116 }
2117
2118 static int mdd_acl_init(const struct lu_env *env, struct mdd_object *pobj,
2119                         struct lu_attr *la, struct lu_buf *def_acl_buf,
2120                         struct lu_buf *acl_buf)
2121 {
2122         int     rc;
2123         ENTRY;
2124
2125         if (S_ISLNK(la->la_mode)) {
2126                 acl_buf->lb_len = 0;
2127                 def_acl_buf->lb_len = 0;
2128                 RETURN(0);
2129         }
2130
2131         mdd_read_lock(env, pobj, MOR_TGT_PARENT);
2132         rc = mdo_xattr_get(env, pobj, def_acl_buf,
2133                            XATTR_NAME_ACL_DEFAULT);
2134         mdd_read_unlock(env, pobj);
2135         if (rc > 0) {
2136                 /* If there are default ACL, fix mode/ACL by default ACL */
2137                 def_acl_buf->lb_len = rc;
2138                 LASSERT(def_acl_buf->lb_len <= acl_buf->lb_len);
2139                 memcpy(acl_buf->lb_buf, def_acl_buf->lb_buf, rc);
2140                 acl_buf->lb_len = rc;
2141                 rc = __mdd_fix_mode_acl(env, acl_buf, &la->la_mode);
2142                 if (rc < 0)
2143                         RETURN(rc);
2144         } else if (rc == -ENODATA || rc == -EOPNOTSUPP) {
2145                 /* If there are no default ACL, fix mode by mask */
2146                 struct lu_ucred *uc = lu_ucred(env);
2147
2148                 /* The create triggered by MDT internal events, such as
2149                  * LFSCK reset, will not contain valid "uc". */
2150                 if (unlikely(uc != NULL))
2151                         la->la_mode &= ~uc->uc_umask;
2152                 rc = 0;
2153                 acl_buf->lb_len = 0;
2154                 def_acl_buf->lb_len = 0;
2155         }
2156
2157         RETURN(rc);
2158 }
2159
2160 /**
2161  * Create a metadata object and initialize it, set acl, xattr.
2162  **/
2163 static int mdd_object_create(const struct lu_env *env, struct mdd_object *pobj,
2164                              struct mdd_object *son, struct lu_attr *attr,
2165                              struct md_op_spec *spec, struct lu_buf *acl_buf,
2166                              struct lu_buf *def_acl_buf,
2167                              struct dt_allocation_hint *hint,
2168                              struct thandle *handle)
2169 {
2170         const struct lu_buf    *buf;
2171         int                     rc;
2172
2173         mdd_write_lock(env, son, MOR_TGT_CHILD);
2174         rc = mdd_object_create_internal(env, NULL, son, attr, handle, spec,
2175                                         hint);
2176         if (rc)
2177                 GOTO(unlock, rc);
2178
2179         /* Note: In DNE phase I, for striped dir, though sub-stripes will be
2180          * created in declare phase, they also needs to be added to master
2181          * object as sub-directory entry. So it has to initialize the master
2182          * object, then set dir striped EA.(in mdo_xattr_set) */
2183         rc = mdd_object_initialize(env, mdo2fid(pobj), son, attr, handle,
2184                                    spec);
2185         if (rc != 0)
2186                 GOTO(err_destroy, rc);
2187
2188         /*
2189          * in case of replay we just set LOVEA provided by the client
2190          * XXX: I think it would be interesting to try "old" way where
2191          *      MDT calls this xattr_set(LOV) in a different transaction.
2192          *      probably this way we code can be made better.
2193          */
2194
2195         /* During creation, there are only a few cases we need do xattr_set to
2196          * create stripes.
2197          * 1. regular file: see comments above.
2198          * 2. dir: inherit default striping or pool settings from parent.
2199          * 3. create striped directory with provided stripeEA.
2200          * 4. create striped directory because inherit default layout from the
2201          * parent.
2202          */
2203         if (spec->no_create ||
2204             (S_ISREG(attr->la_mode) && spec->sp_cr_flags & MDS_OPEN_HAS_EA) ||
2205             S_ISDIR(attr->la_mode)) {
2206                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2207                                         spec->u.sp_ea.eadatalen);
2208                 rc = mdo_xattr_set(env, son, buf,
2209                                    S_ISDIR(attr->la_mode) ? XATTR_NAME_LMV :
2210                                                             XATTR_NAME_LOV, 0,
2211                                    handle);
2212                 if (rc != 0)
2213                         GOTO(err_destroy, rc);
2214         }
2215
2216 #ifdef CONFIG_FS_POSIX_ACL
2217         if (def_acl_buf != NULL && def_acl_buf->lb_len > 0 &&
2218             S_ISDIR(attr->la_mode)) {
2219                 /* set default acl */
2220                 rc = mdo_xattr_set(env, son, def_acl_buf,
2221                                    XATTR_NAME_ACL_DEFAULT, 0,
2222                                    handle);
2223                 if (rc)
2224                         GOTO(err_destroy, rc);
2225         }
2226         /* set its own acl */
2227         if (acl_buf != NULL && acl_buf->lb_len > 0) {
2228                 rc = mdo_xattr_set(env, son, acl_buf,
2229                                    XATTR_NAME_ACL_ACCESS,
2230                                    0, handle);
2231                 if (rc)
2232                         GOTO(err_destroy, rc);
2233         }
2234 #endif
2235
2236         if (S_ISLNK(attr->la_mode)) {
2237                 struct lu_ucred  *uc = lu_ucred_assert(env);
2238                 struct dt_object *dt = mdd_object_child(son);
2239                 const char *target_name = spec->u.sp_symname;
2240                 int sym_len = strlen(target_name);
2241                 const struct lu_buf *buf;
2242                 loff_t pos = 0;
2243
2244                 buf = mdd_buf_get_const(env, target_name, sym_len);
2245                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
2246                                                 uc->uc_cap &
2247                                                 CFS_CAP_SYS_RESOURCE_MASK);
2248
2249                 if (rc == sym_len)
2250                         rc = 0;
2251                 else
2252                         GOTO(err_initlized, rc = -EFAULT);
2253         }
2254
2255         if (spec->sp_cr_file_secctx_name != NULL) {
2256                 buf = mdd_buf_get_const(env, spec->sp_cr_file_secctx,
2257                                         spec->sp_cr_file_secctx_size);
2258                 rc = mdo_xattr_set(env, son, buf, spec->sp_cr_file_secctx_name,
2259                                    0, handle);
2260                 if (rc < 0)
2261                         GOTO(err_initlized, rc);
2262         }
2263
2264 err_initlized:
2265         if (unlikely(rc != 0)) {
2266                 int rc2;
2267                 if (S_ISDIR(attr->la_mode)) {
2268                         /* Drop the reference, no need to delete "."/"..",
2269                          * because the object to be destroied directly. */
2270                         rc2 = mdo_ref_del(env, son, handle);
2271                         if (rc2 != 0)
2272                                 GOTO(unlock, rc);
2273                 }
2274                 rc2 = mdo_ref_del(env, son, handle);
2275                 if (rc2 != 0)
2276                         GOTO(unlock, rc);
2277 err_destroy:
2278                 mdo_destroy(env, son, handle);
2279         }
2280 unlock:
2281         mdd_write_unlock(env, son);
2282         RETURN(rc);
2283 }
2284
2285 static int mdd_index_delete(const struct lu_env *env,
2286                             struct mdd_object *mdd_pobj,
2287                             struct lu_attr *cattr,
2288                             const struct lu_name *lname)
2289 {
2290         struct mdd_device *mdd = mdo2mdd(&mdd_pobj->mod_obj);
2291         struct thandle *handle;
2292         int rc;
2293         ENTRY;
2294
2295         handle = mdd_trans_create(env, mdd);
2296         if (IS_ERR(handle))
2297                 RETURN(PTR_ERR(handle));
2298
2299         rc = mdo_declare_index_delete(env, mdd_pobj, lname->ln_name,
2300                                       handle);
2301         if (rc != 0)
2302                 GOTO(stop, rc);
2303
2304         if (S_ISDIR(cattr->la_mode)) {
2305                 rc = mdo_declare_ref_del(env, mdd_pobj, handle);
2306                 if (rc != 0)
2307                         GOTO(stop, rc);
2308         }
2309
2310         /* Since this will only be used in the error handler path,
2311          * Let's set the thandle to be local and not mess the transno */
2312         handle->th_local = 1;
2313         rc = mdd_trans_start(env, mdd, handle);
2314         if (rc)
2315                 GOTO(stop, rc);
2316
2317         rc = __mdd_index_delete(env, mdd_pobj, lname->ln_name,
2318                                 S_ISDIR(cattr->la_mode), handle);
2319         if (rc)
2320                 GOTO(stop, rc);
2321 stop:
2322         rc = mdd_trans_stop(env, mdd, rc, handle);
2323
2324         RETURN(rc);
2325 }
2326
2327 /*
2328  * Create object and insert it into namespace.
2329  */
2330 static int mdd_create(const struct lu_env *env, struct md_object *pobj,
2331                       const struct lu_name *lname, struct md_object *child,
2332                       struct md_op_spec *spec, struct md_attr* ma)
2333 {
2334         struct mdd_thread_info  *info = mdd_env_info(env);
2335         struct lu_attr          *la = &info->mti_la_for_fix;
2336         struct mdd_object       *mdd_pobj = md2mdd_obj(pobj);
2337         struct mdd_object       *son = md2mdd_obj(child);
2338         struct mdd_device       *mdd = mdo2mdd(pobj);
2339         struct lu_attr          *attr = &ma->ma_attr;
2340         struct thandle          *handle;
2341         struct lu_attr          *pattr = &info->mti_pattr;
2342         struct lu_buf           acl_buf;
2343         struct lu_buf           def_acl_buf;
2344         struct linkea_data      *ldata = &info->mti_link_data;
2345         const char              *name = lname->ln_name;
2346         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
2347         int                      rc;
2348         int                      rc2;
2349         ENTRY;
2350
2351         /*
2352          * Two operations have to be performed:
2353          *
2354          *  - an allocation of a new object (->do_create()), and
2355          *
2356          *  - an insertion into a parent index (->dio_insert()).
2357          *
2358          * Due to locking, operation order is not important, when both are
2359          * successful, *but* error handling cases are quite different:
2360          *
2361          *  - if insertion is done first, and following object creation fails,
2362          *  insertion has to be rolled back, but this operation might fail
2363          *  also leaving us with dangling index entry.
2364          *
2365          *  - if creation is done first, is has to be undone if insertion
2366          *  fails, leaving us with leaked space, which is neither good, nor
2367          *  fatal.
2368          *
2369          * It seems that creation-first is simplest solution, but it is
2370          * sub-optimal in the frequent
2371          *
2372          *         $ mkdir foo
2373          *         $ mkdir foo
2374          *
2375          * case, because second mkdir is bound to create object, only to
2376          * destroy it immediately.
2377          *
2378          * To avoid this follow local file systems that do double lookup:
2379          *
2380          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
2381          *
2382          *     1. create            (mdd_object_create_internal())
2383          *
2384          *     2. insert            (__mdd_index_insert(), lookup again)
2385          */
2386
2387         rc = mdd_la_get(env, mdd_pobj, pattr);
2388         if (rc != 0)
2389                 RETURN(rc);
2390
2391         /* Sanity checks before big job. */
2392         rc = mdd_create_sanity_check(env, pobj, pattr, lname, attr, spec);
2393         if (rc)
2394                 RETURN(rc);
2395
2396         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_DQACQ_NET))
2397                 GOTO(out_free, rc = -EINPROGRESS);
2398
2399         handle = mdd_trans_create(env, mdd);
2400         if (IS_ERR(handle))
2401                 GOTO(out_free, rc = PTR_ERR(handle));
2402
2403         acl_buf.lb_buf = info->mti_xattr_buf;
2404         acl_buf.lb_len = sizeof(info->mti_xattr_buf);
2405         def_acl_buf.lb_buf = info->mti_key;
2406         def_acl_buf.lb_len = sizeof(info->mti_key);
2407         rc = mdd_acl_init(env, mdd_pobj, attr, &def_acl_buf, &acl_buf);
2408         if (rc < 0)
2409                 GOTO(out_stop, rc);
2410
2411         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
2412
2413         memset(ldata, 0, sizeof(*ldata));
2414         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
2415                 struct lu_fid tfid = *mdd_object_fid(mdd_pobj);
2416
2417                 tfid.f_oid--;
2418                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2419                                         &tfid, lname, 1, 0, ldata);
2420         } else {
2421                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2422                                         mdd_object_fid(mdd_pobj),
2423                                         lname, 1, 0, ldata);
2424         }
2425
2426         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, attr,
2427                                 handle, spec, ldata, &def_acl_buf, &acl_buf,
2428                                 hint);
2429         if (rc)
2430                 GOTO(out_stop, rc);
2431
2432         rc = mdd_trans_start(env, mdd, handle);
2433         if (rc)
2434                 GOTO(out_stop, rc);
2435
2436         rc = mdd_object_create(env, mdd_pobj, son, attr, spec, &acl_buf,
2437                                &def_acl_buf, hint, handle);
2438         if (rc != 0)
2439                 GOTO(out_stop, rc);
2440
2441         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2442                 mdd_write_lock(env, son, MOR_TGT_CHILD);
2443                 rc = __mdd_orphan_add(env, son, handle);
2444                 GOTO(out_volatile, rc);
2445         } else {
2446                 rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2447                                         attr->la_mode, name, handle);
2448                 if (rc != 0)
2449                         GOTO(err_created, rc);
2450
2451                 mdd_links_add(env, son, mdo2fid(mdd_pobj), lname, handle,
2452                               ldata, 1);
2453
2454                 /* update parent directory mtime/ctime */
2455                 *la = *attr;
2456                 la->la_valid = LA_CTIME | LA_MTIME;
2457                 rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
2458                 if (rc)
2459                         GOTO(err_insert, rc);
2460         }
2461
2462         EXIT;
2463 err_insert:
2464         if (rc != 0) {
2465                 if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
2466                         rc2 = __mdd_orphan_del(env, son, handle);
2467                 else
2468                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2469                                                  S_ISDIR(attr->la_mode),
2470                                                  handle);
2471                 if (rc2 != 0)
2472                         goto out_stop;
2473
2474 err_created:
2475                 mdd_write_lock(env, son, MOR_TGT_CHILD);
2476                 if (S_ISDIR(attr->la_mode)) {
2477                         /* Drop the reference, no need to delete "."/"..",
2478                          * because the object is to be destroyed directly. */
2479                         rc2 = mdo_ref_del(env, son, handle);
2480                         if (rc2 != 0) {
2481                                 mdd_write_unlock(env, son);
2482                                 goto out_stop;
2483                         }
2484                 }
2485 out_volatile:
2486                 /* For volatile files drop one link immediately, since there is
2487                  * no filename in the namespace, and save any error returned. */
2488                 rc2 = mdo_ref_del(env, son, handle);
2489                 if (rc2 != 0) {
2490                         mdd_write_unlock(env, son);
2491                         if (unlikely(rc == 0))
2492                                 rc = rc2;
2493                         goto out_stop;
2494                 }
2495
2496                 /* Don't destroy the volatile object on success */
2497                 if (likely(rc != 0))
2498                         mdo_destroy(env, son, handle);
2499                 mdd_write_unlock(env, son);
2500         }
2501
2502         if (rc == 0 && fid_is_namespace_visible(mdo2fid(son)) &&
2503             likely((spec->sp_cr_flags & MDS_OPEN_VOLATILE) == 0))
2504                 rc = mdd_changelog_ns_store(env, mdd,
2505                                 S_ISDIR(attr->la_mode) ? CL_MKDIR :
2506                                 S_ISREG(attr->la_mode) ? CL_CREATE :
2507                                 S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2508                                 0, son, mdo2fid(mdd_pobj), NULL, NULL, lname,
2509                                 NULL, handle);
2510 out_stop:
2511         rc2 = mdd_trans_stop(env, mdd, rc, handle);
2512         if (rc == 0) {
2513                 /* If creation fails, it is most likely due to the remote update
2514                  * failure, because local transaction will mostly succeed at
2515                  * this stage. There is no easy way to rollback all of previous
2516                  * updates, so let's remove the object from namespace, and
2517                  * LFSCK should handle the orphan object. */
2518                 if (rc2 < 0 && !mdd_object_remote(mdd_pobj))
2519                         mdd_index_delete(env, mdd_pobj, attr, lname);
2520                 rc = rc2;
2521         }
2522 out_free:
2523         if (is_vmalloc_addr(ldata->ld_buf))
2524                 /* if we vmalloced a large buffer drop it */
2525                 lu_buf_free(ldata->ld_buf);
2526
2527         /* The child object shouldn't be cached anymore */
2528         if (rc)
2529                 set_bit(LU_OBJECT_HEARD_BANSHEE,
2530                         &child->mo_lu.lo_header->loh_flags);
2531         return rc;
2532 }
2533
2534 /*
2535  * Get locks on parents in proper order
2536  * RETURN: < 0 - error, rename_order if successful
2537  */
2538 enum rename_order {
2539         MDD_RN_SAME,
2540         MDD_RN_SRCTGT,
2541         MDD_RN_TGTSRC
2542 };
2543
2544 static int mdd_rename_order(const struct lu_env *env,
2545                             struct mdd_device *mdd,
2546                             struct mdd_object *src_pobj,
2547                             const struct lu_attr *pattr,
2548                             struct mdd_object *tgt_pobj)
2549 {
2550         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2551         int rc;
2552         ENTRY;
2553
2554         if (src_pobj == tgt_pobj)
2555                 RETURN(MDD_RN_SAME);
2556
2557         /* compared the parent child relationship of src_p&tgt_p */
2558         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2559                 rc = MDD_RN_SRCTGT;
2560         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2561                 rc = MDD_RN_TGTSRC;
2562         } else {
2563                 rc = mdd_is_parent(env, mdd, src_pobj, pattr, mdo2fid(tgt_pobj),
2564                                    NULL);
2565                 if (rc == -EREMOTE)
2566                         rc = 0;
2567
2568                 if (rc == 1)
2569                         rc = MDD_RN_TGTSRC;
2570                 else if (rc == 0)
2571                         rc = MDD_RN_SRCTGT;
2572         }
2573
2574         RETURN(rc);
2575 }
2576
2577 /* has not mdd_write{read}_lock on any obj yet. */
2578 static int mdd_rename_sanity_check(const struct lu_env *env,
2579                                    struct mdd_object *src_pobj,
2580                                    const struct lu_attr *pattr,
2581                                    struct mdd_object *tgt_pobj,
2582                                    const struct lu_attr *tpattr,
2583                                    struct mdd_object *sobj,
2584                                    const struct lu_attr *cattr,
2585                                    struct mdd_object *tobj,
2586                                    const struct lu_attr *tattr)
2587 {
2588         int rc = 0;
2589         ENTRY;
2590
2591         /* XXX: when get here, sobj must NOT be NULL,
2592          * the other case has been processed in cld_rename
2593          * before mdd_rename and enable MDS_PERM_BYPASS. */
2594         LASSERT(sobj);
2595
2596         rc = mdd_may_delete(env, src_pobj, pattr, sobj, cattr, NULL, 1, 0);
2597         if (rc)
2598                 RETURN(rc);
2599
2600         /* XXX: when get here, "tobj == NULL" means tobj must
2601          * NOT exist (neither on remote MDS, such case has been
2602          * processed in cld_rename before mdd_rename and enable
2603          * MDS_PERM_BYPASS).
2604          * So check may_create, but not check may_unlink. */
2605         if (tobj == NULL)
2606                 rc = mdd_may_create(env, tgt_pobj, tpattr, NULL,
2607                                     (src_pobj != tgt_pobj));
2608         else
2609                 rc = mdd_may_delete(env, tgt_pobj, tpattr, tobj, tattr, cattr,
2610                                     (src_pobj != tgt_pobj), 1);
2611
2612         if (!rc && !tobj && (src_pobj != tgt_pobj) && S_ISDIR(cattr->la_mode))
2613                 rc = __mdd_may_link(env, tgt_pobj, tpattr);
2614
2615         RETURN(rc);
2616 }
2617
2618 static int mdd_declare_rename(const struct lu_env *env,
2619                               struct mdd_device *mdd,
2620                               struct mdd_object *mdd_spobj,
2621                               struct mdd_object *mdd_tpobj,
2622                               struct mdd_object *mdd_sobj,
2623                               struct mdd_object *mdd_tobj,
2624                               const struct lu_name *sname,
2625                               const struct lu_name *tname,
2626                               struct md_attr *ma,
2627                               struct linkea_data *ldata,
2628                               struct thandle *handle)
2629 {
2630         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2631         int rc;
2632
2633         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2634         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2635
2636         LASSERT(mdd_spobj);
2637         LASSERT(mdd_tpobj);
2638         LASSERT(mdd_sobj);
2639
2640         /* name from source dir */
2641         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2642         if (rc)
2643                 return rc;
2644
2645         /* .. from source child */
2646         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2647                 /* source child can be directory,
2648                  * counted by source dir's nlink */
2649                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2650                 if (rc)
2651                         return rc;
2652                 if (mdd_spobj != mdd_tpobj) {
2653                         rc = mdo_declare_index_delete(env, mdd_sobj, dotdot,
2654                                                       handle);
2655                         if (rc != 0)
2656                                 return rc;
2657
2658                         rc = mdo_declare_index_insert(env, mdd_sobj,
2659                                                       mdo2fid(mdd_tpobj),
2660                                                       S_IFDIR, dotdot, handle);
2661                         if (rc != 0)
2662                                 return rc;
2663                 }
2664
2665                 /* new target child can be directory,
2666                  * counted by target dir's nlink */
2667                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2668                 if (rc != 0)
2669                         return rc;
2670         }
2671
2672         la->la_valid = LA_CTIME | LA_MTIME;
2673         rc = mdo_declare_attr_set(env, mdd_spobj, la, handle);
2674         if (rc != 0)
2675                 return rc;
2676
2677         rc = mdo_declare_attr_set(env, mdd_tpobj, la, handle);
2678         if (rc != 0)
2679                 return rc;
2680
2681         la->la_valid = LA_CTIME;
2682         rc = mdo_declare_attr_set(env, mdd_sobj, la, handle);
2683         if (rc)
2684                 return rc;
2685
2686         rc = mdd_declare_links_add(env, mdd_sobj, handle, ldata);
2687         if (rc)
2688                 return rc;
2689
2690         /* new name */
2691         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2692                                       mdd_object_type(mdd_sobj),
2693                                       tname->ln_name, handle);
2694         if (rc != 0)
2695                 return rc;
2696
2697         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2698                 /* delete target child in target parent directory */
2699                 rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name,
2700                                               handle);
2701                 if (rc)
2702                         return rc;
2703
2704                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2705                 if (rc)
2706                         return rc;
2707
2708                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2709                         /* target child can be directory,
2710                          * delete "." reference in target child directory */
2711                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2712                         if (rc)
2713                                 return rc;
2714
2715                         /* delete ".." reference in target parent directory */
2716                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2717                         if (rc)
2718                                 return rc;
2719                 }
2720
2721                 la->la_valid = LA_CTIME;
2722                 rc = mdo_declare_attr_set(env, mdd_tobj, la, handle);
2723                 if (rc)
2724                         return rc;
2725
2726                 rc = mdd_declare_finish_unlink(env, mdd_tobj, handle);
2727                 if (rc)
2728                         return rc;
2729         }
2730
2731         rc = mdd_declare_changelog_store(env, mdd, tname, sname, handle);
2732         if (rc)
2733                 return rc;
2734
2735         return rc;
2736 }
2737
2738 /* src object can be remote that is why we use only fid and type of object */
2739 static int mdd_rename(const struct lu_env *env,
2740                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2741                       const struct lu_fid *lf, const struct lu_name *lsname,
2742                       struct md_object *tobj, const struct lu_name *ltname,
2743                       struct md_attr *ma)
2744 {
2745         const char *sname = lsname->ln_name;
2746         const char *tname = ltname->ln_name;
2747         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2748         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2749         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2750         struct mdd_device *mdd = mdo2mdd(src_pobj);
2751         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2752         struct mdd_object *mdd_tobj = NULL;
2753         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
2754         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
2755         struct lu_attr *tattr = MDD_ENV_VAR(env, tattr);
2756         struct lu_attr *tpattr = MDD_ENV_VAR(env, tpattr);
2757         struct thandle *handle;
2758         struct linkea_data  *ldata = &mdd_env_info(env)->mti_link_data;
2759         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2760         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2761         bool is_dir;
2762         bool tobj_ref = 0;
2763         bool tobj_locked = 0;
2764         unsigned cl_flags = 0;
2765         int rc, rc2;
2766         ENTRY;
2767
2768         if (tobj)
2769                 mdd_tobj = md2mdd_obj(tobj);
2770
2771         mdd_sobj = mdd_object_find(env, mdd, lf);
2772         if (IS_ERR(mdd_sobj))
2773                 RETURN(PTR_ERR(mdd_sobj));
2774
2775         rc = mdd_la_get(env, mdd_sobj, cattr);
2776         if (rc)
2777                 GOTO(out_pending, rc);
2778
2779         rc = mdd_la_get(env, mdd_spobj, pattr);
2780         if (rc)
2781                 GOTO(out_pending, rc);
2782
2783         if (mdd_tobj) {
2784                 rc = mdd_la_get(env, mdd_tobj, tattr);
2785                 if (rc)
2786                         GOTO(out_pending, rc);
2787                 /* search for an existing archive.
2788                  * we should check ahead as the object
2789                  * can be destroyed in this transaction */
2790                 if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
2791                         cl_flags |= CLF_RENAME_LAST_EXISTS;
2792         }
2793
2794         rc = mdd_la_get(env, mdd_tpobj, tpattr);
2795         if (rc)
2796                 GOTO(out_pending, rc);
2797
2798         rc = mdd_rename_sanity_check(env, mdd_spobj, pattr, mdd_tpobj, tpattr,
2799                                      mdd_sobj, cattr, mdd_tobj, tattr);
2800         if (rc)
2801                 GOTO(out_pending, rc);
2802
2803         rc = mdd_name_check(mdd, ltname);
2804         if (rc < 0)
2805                 GOTO(out_pending, rc);
2806
2807         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2808         rc = mdd_rename_order(env, mdd, mdd_spobj, pattr, mdd_tpobj);
2809         if (rc < 0)
2810                 GOTO(out_pending, rc);
2811
2812         handle = mdd_trans_create(env, mdd);
2813         if (IS_ERR(handle))
2814                 GOTO(out_pending, rc = PTR_ERR(handle));
2815
2816         memset(ldata, 0, sizeof(*ldata));
2817         rc = mdd_linkea_prepare(env, mdd_sobj, mdd_object_fid(mdd_spobj),
2818                                 lsname, mdd_object_fid(mdd_tpobj), ltname,
2819                                 1, 0, ldata);
2820         if (rc)
2821                 GOTO(stop, rc);
2822
2823         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
2824                                 mdd_tobj, lsname, ltname, ma, ldata, handle);
2825         if (rc)
2826                 GOTO(stop, rc);
2827
2828         rc = mdd_trans_start(env, mdd, handle);
2829         if (rc)
2830                 GOTO(stop, rc);
2831
2832         is_dir = S_ISDIR(cattr->la_mode);
2833
2834         /* Remove source name from source directory */
2835         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle);
2836         if (rc != 0)
2837                 GOTO(stop, rc);
2838
2839         /* "mv dir1 dir2" needs "dir1/.." link update */
2840         if (is_dir && !lu_fid_eq(spobj_fid, tpobj_fid)) {
2841                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
2842                 if (rc != 0)
2843                         GOTO(fixup_spobj2, rc);
2844
2845                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, S_IFDIR,
2846                                              dotdot, handle);
2847                 if (rc != 0)
2848                         GOTO(fixup_spobj, rc);
2849         }
2850
2851         if (mdd_tobj != NULL && mdd_object_exists(mdd_tobj)) {
2852                 rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
2853                 if (rc != 0)
2854                         /* tname might been renamed to something else */
2855                         GOTO(fixup_spobj, rc);
2856         }
2857
2858         /* Insert new fid with target name into target dir */
2859         rc = __mdd_index_insert(env, mdd_tpobj, lf, cattr->la_mode,
2860                                 tname, handle);
2861         if (rc != 0)
2862                 GOTO(fixup_tpobj, rc);
2863
2864         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2865         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2866
2867         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
2868         la->la_valid = LA_CTIME;
2869         rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
2870         if (rc)
2871                 GOTO(fixup_tpobj, rc);
2872
2873         /* Update the linkEA for the source object */
2874         mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
2875         rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
2876                               mdo2fid(mdd_tpobj), ltname, handle, ldata,
2877                               0, 0);
2878         if (rc == -ENOENT)
2879                 /* Old files might not have EA entry */
2880                 mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
2881                               lsname, handle, NULL, 0);
2882         mdd_write_unlock(env, mdd_sobj);
2883         /* We don't fail the transaction if the link ea can't be
2884            updated -- fid2path will use alternate lookup method. */
2885         rc = 0;
2886
2887         /* Remove old target object
2888          * For tobj is remote case cmm layer has processed
2889          * and set tobj to NULL then. So when tobj is NOT NULL,
2890          * it must be local one.
2891          */
2892         if (tobj && mdd_object_exists(mdd_tobj)) {
2893                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
2894                 tobj_locked = 1;
2895                 if (mdd_is_dead_obj(mdd_tobj)) {
2896                         /* shld not be dead, something is wrong */
2897                         CERROR("tobj is dead, something is wrong\n");
2898                         rc = -EINVAL;
2899                         goto cleanup;
2900                 }
2901                 mdo_ref_del(env, mdd_tobj, handle);
2902
2903                 /* Remove dot reference. */
2904                 if (S_ISDIR(tattr->la_mode))
2905                         mdo_ref_del(env, mdd_tobj, handle);
2906                 tobj_ref = 1;
2907
2908                 /* fetch updated nlink */
2909                 rc = mdd_la_get(env, mdd_tobj, tattr);
2910                 if (rc != 0) {
2911                         CERROR("%s: Failed to get nlink for tobj "
2912                                 DFID": rc = %d\n",
2913                                 mdd2obd_dev(mdd)->obd_name,
2914                                 PFID(tpobj_fid), rc);
2915                         GOTO(fixup_tpobj, rc);
2916                 }
2917
2918                 la->la_valid = LA_CTIME;
2919                 rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
2920                 if (rc != 0) {
2921                         CERROR("%s: Failed to set ctime for tobj "
2922                                 DFID": rc = %d\n",
2923                                 mdd2obd_dev(mdd)->obd_name,
2924                                 PFID(tpobj_fid), rc);
2925                         GOTO(fixup_tpobj, rc);
2926                 }
2927
2928                 /* XXX: this transfer to ma will be removed with LOD/OSP */
2929                 ma->ma_attr = *tattr;
2930                 ma->ma_valid |= MA_INODE;
2931                 rc = mdd_finish_unlink(env, mdd_tobj, ma, mdd_tpobj, ltname,
2932                                        handle);
2933                 if (rc != 0) {
2934                         CERROR("%s: Failed to unlink tobj "
2935                                 DFID": rc = %d\n",
2936                                 mdd2obd_dev(mdd)->obd_name,
2937                                 PFID(tpobj_fid), rc);
2938                         GOTO(fixup_tpobj, rc);
2939                 }
2940
2941                 /* fetch updated nlink */
2942                 rc = mdd_la_get(env, mdd_tobj, tattr);
2943                 if (rc == -ENOENT) {
2944                         /* the object got removed, let's
2945                          * return the latest known attributes */
2946                         tattr->la_nlink = 0;
2947                         rc = 0;
2948                 } else if (rc != 0) {
2949                         CERROR("%s: Failed to get nlink for tobj "
2950                                 DFID": rc = %d\n",
2951                                 mdd2obd_dev(mdd)->obd_name,
2952                                 PFID(tpobj_fid), rc);
2953                         GOTO(fixup_tpobj, rc);
2954                 }
2955                 /* XXX: this transfer to ma will be removed with LOD/OSP */
2956                 ma->ma_attr = *tattr;
2957                 ma->ma_valid |= MA_INODE;
2958
2959                 if (tattr->la_nlink == 0)
2960                         cl_flags |= CLF_RENAME_LAST;
2961                 else
2962                         cl_flags &= ~CLF_RENAME_LAST_EXISTS;
2963         }
2964
2965         la->la_valid = LA_CTIME | LA_MTIME;
2966         rc = mdd_update_time(env, mdd_spobj, pattr, la, handle);
2967         if (rc)
2968                 GOTO(fixup_tpobj, rc);
2969
2970         if (mdd_spobj != mdd_tpobj) {
2971                 la->la_valid = LA_CTIME | LA_MTIME;
2972                 rc = mdd_update_time(env, mdd_tpobj, tpattr, la, handle);
2973                 if (rc != 0)
2974                         GOTO(fixup_tpobj, rc);
2975         }
2976
2977         EXIT;
2978
2979 fixup_tpobj:
2980         if (rc) {
2981                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
2982                 if (rc2)
2983                         CWARN("tp obj fix error %d\n",rc2);
2984
2985                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
2986                     !mdd_is_dead_obj(mdd_tobj)) {
2987                         if (tobj_ref) {
2988                                 mdo_ref_add(env, mdd_tobj, handle);
2989                                 if (is_dir)
2990                                         mdo_ref_add(env, mdd_tobj, handle);
2991                         }
2992
2993                         rc2 = __mdd_index_insert(env, mdd_tpobj,
2994                                                   mdo2fid(mdd_tobj),
2995                                                   mdd_object_type(mdd_tobj),
2996                                                   tname, handle);
2997                         if (rc2 != 0)
2998                                 CWARN("tp obj fix error: rc = %d\n", rc2);
2999                 }
3000         }
3001
3002 fixup_spobj:
3003         if (rc && is_dir && mdd_sobj && mdd_spobj != mdd_tpobj) {
3004                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3005                 if (rc2)
3006                         CWARN("%s: sp obj dotdot delete error: rc = %d\n",
3007                                mdd2obd_dev(mdd)->obd_name, rc2);
3008
3009
3010                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid, S_IFDIR,
3011                                               dotdot, handle);
3012                 if (rc2 != 0)
3013                         CWARN("%s: sp obj dotdot insert error: rc = %d\n",
3014                               mdd2obd_dev(mdd)->obd_name, rc2);
3015         }
3016
3017 fixup_spobj2:
3018         if (rc != 0) {
3019                 rc2 = __mdd_index_insert(env, mdd_spobj, lf,
3020                                          mdd_object_type(mdd_sobj), sname,
3021                                          handle);
3022                 if (rc2 != 0)
3023                         CWARN("sp obj fix error: rc = %d\n", rc2);
3024         }
3025
3026 cleanup:
3027         if (tobj_locked)
3028                 mdd_write_unlock(env, mdd_tobj);
3029
3030         if (rc == 0)
3031                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, cl_flags,
3032                                             mdd_tobj, tpobj_fid, lf, spobj_fid,
3033                                             ltname, lsname, handle);
3034
3035 stop:
3036         rc = mdd_trans_stop(env, mdd, rc, handle);
3037
3038 out_pending:
3039         mdd_object_put(env, mdd_sobj);
3040         return rc;
3041 }
3042
3043 /**
3044  * During migration once the parent FID has been changed,
3045  * we need update the parent FID in linkea.
3046  **/
3047 static int mdd_linkea_update_child_internal(const struct lu_env *env,
3048                                             struct mdd_object *parent,
3049                                             struct mdd_object *newparent,
3050                                             struct mdd_object *child,
3051                                             const char *name, int namelen,
3052                                             struct thandle *handle,
3053                                             bool declare)
3054 {
3055         struct mdd_thread_info  *info = mdd_env_info(env);
3056         struct linkea_data      ldata = { NULL };
3057         struct lu_buf           *buf = &info->mti_link_buf;
3058         int                     count;
3059         int                     rc = 0;
3060
3061         ENTRY;
3062
3063         buf = lu_buf_check_and_alloc(buf, PATH_MAX);
3064         if (buf->lb_buf == NULL)
3065                 RETURN(-ENOMEM);
3066
3067         ldata.ld_buf = buf;
3068         rc = mdd_links_read(env, child, &ldata);
3069         if (rc != 0) {
3070                 if (rc == -ENOENT || rc == -ENODATA)
3071                         rc = 0;
3072                 RETURN(rc);
3073         }
3074
3075         LASSERT(ldata.ld_leh != NULL);
3076         ldata.ld_lee = (struct link_ea_entry *)(ldata.ld_leh + 1);
3077         for (count = 0; count < ldata.ld_leh->leh_reccount; count++) {
3078                 struct mdd_device *mdd = mdo2mdd(&child->mod_obj);
3079                 struct lu_name lname;
3080                 struct lu_fid  fid;
3081
3082                 linkea_entry_unpack(ldata.ld_lee, &ldata.ld_reclen,
3083                                     &lname, &fid);
3084
3085                 if (strncmp(lname.ln_name, name, namelen) != 0 ||
3086                     !lu_fid_eq(&fid, mdd_object_fid(parent))) {
3087                         ldata.ld_lee = (struct link_ea_entry *)
3088                                        ((char *)ldata.ld_lee +
3089                                         ldata.ld_reclen);
3090                         continue;
3091                 }
3092
3093                 CDEBUG(D_INFO, "%s: update "DFID" with %.*s:"DFID"\n",
3094                        mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(child)),
3095                        lname.ln_namelen, lname.ln_name,
3096                        PFID(mdd_object_fid(newparent)));
3097                 /* update to the new parent fid */
3098                 linkea_entry_pack(ldata.ld_lee, &lname,
3099                                   mdd_object_fid(newparent));
3100                 if (declare)
3101                         rc = mdd_declare_links_add(env, child, handle, &ldata);
3102                 else
3103                         rc = mdd_links_write(env, child, &ldata, handle);
3104                 break;
3105         }
3106         RETURN(rc);
3107 }
3108
3109 static int mdd_linkea_declare_update_child(const struct lu_env *env,
3110                                            struct mdd_object *parent,
3111                                            struct mdd_object *newparent,
3112                                            struct mdd_object *child,
3113                                            const char *name, int namelen,
3114                                            struct thandle *handle)
3115 {
3116         return mdd_linkea_update_child_internal(env, parent, newparent,
3117                                                 child, name,
3118                                                 namelen, handle, true);
3119 }
3120
3121 static int mdd_linkea_update_child(const struct lu_env *env,
3122                                    struct mdd_object *parent,
3123                                    struct mdd_object *newparent,
3124                                    struct mdd_object *child,
3125                                    const char *name, int namelen,
3126                                    struct thandle *handle)
3127 {
3128         return mdd_linkea_update_child_internal(env, parent, newparent,
3129                                                 child, name,
3130                                                 namelen, handle, false);
3131 }
3132
3133 static int mdd_update_linkea_internal(const struct lu_env *env,
3134                                       struct mdd_object *mdd_pobj,
3135                                       struct mdd_object *mdd_sobj,
3136                                       struct mdd_object *mdd_tobj,
3137                                       const struct lu_name *child_name,
3138                                       struct linkea_data *ldata,
3139                                       struct thandle *handle,
3140                                       int declare)
3141 {
3142         struct mdd_thread_info  *info = mdd_env_info(env);
3143         int                     count;
3144         int                     rc = 0;
3145         ENTRY;
3146
3147         LASSERT(ldata->ld_buf != NULL);
3148         LASSERT(ldata->ld_leh != NULL);
3149
3150         /* If it is mulitple links file, we need update the name entry for
3151          * all parent */
3152         ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
3153         for (count = 0; count < ldata->ld_leh->leh_reccount; count++) {
3154                 struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3155                 struct mdd_object       *pobj;
3156                 struct lu_name          lname;
3157                 struct lu_fid           fid;
3158
3159                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
3160                                     &lname, &fid);
3161                 pobj = mdd_object_find(env, mdd, &fid);
3162                 if (IS_ERR(pobj)) {
3163                         CWARN("%s: cannot find obj "DFID": rc = %ld\n",
3164                               mdd2obd_dev(mdd)->obd_name, PFID(&fid),
3165                               PTR_ERR(pobj));
3166                         continue;
3167                 }
3168
3169                 if (!mdd_object_exists(pobj)) {
3170                         CDEBUG(D_INFO, "%s: obj "DFID" does not exist\n",
3171                               mdd2obd_dev(mdd)->obd_name, PFID(&fid));
3172                         goto next_put;
3173                 }
3174
3175                 if (pobj == mdd_pobj &&
3176                     lname.ln_namelen == child_name->ln_namelen &&
3177                     strncmp(lname.ln_name, child_name->ln_name,
3178                             lname.ln_namelen) == 0) {
3179                         CDEBUG(D_INFO, "%s: skip its own %s: "DFID"\n",
3180                               mdd2obd_dev(mdd)->obd_name, child_name->ln_name,
3181                               PFID(&fid));
3182                         goto next_put;
3183                 }
3184
3185                 CDEBUG(D_INFO, "%s: update "DFID" with "DNAME":"DFID"\n",
3186                        mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(pobj)),
3187                        PNAME(&lname), PFID(mdd_object_fid(mdd_tobj)));
3188
3189                 if (declare) {
3190                         /* Remove source name from source directory */
3191                         /* Insert new fid with target name into target dir */
3192                         rc = mdo_declare_index_delete(env, pobj, lname.ln_name,
3193                                                       handle);
3194                         if (rc != 0)
3195                                 GOTO(next_put, rc);
3196
3197                         rc = mdo_declare_index_insert(env, pobj,
3198                                         mdd_object_fid(mdd_tobj),
3199                                         mdd_object_type(mdd_tobj),
3200                                         lname.ln_name, handle);
3201                         if (rc != 0)
3202                                 GOTO(next_put, rc);
3203
3204                         rc = mdo_declare_ref_add(env, mdd_tobj, handle);
3205                         if (rc)
3206                                 GOTO(next_put, rc);
3207
3208                         rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3209                         if (rc)
3210                                 GOTO(next_put, rc);
3211                 } else {
3212                         char *tmp_name = info->mti_key;
3213
3214                         if (lname.ln_namelen >= sizeof(info->mti_key)) {
3215                                 /* lnamelen is too big(> NAME_MAX + 16),
3216                                  * something wrong about this linkea, let's
3217                                  * skip it */
3218                                 CWARN("%s: the name %.*s is too long under "
3219                                       DFID"\n", mdd2obd_dev(mdd)->obd_name,
3220                                       lname.ln_namelen, lname.ln_name,
3221                                       PFID(&fid));
3222                                 goto next_put;
3223                         }
3224
3225                         /* Note: lname might be without \0 at the end, see
3226                          * linkea_entry_unpack(), let's add extra \0 by
3227                          * snprintf */
3228                         snprintf(tmp_name, sizeof(info->mti_key), "%.*s",
3229                                  lname.ln_namelen, lname.ln_name);
3230                         lname.ln_name = tmp_name;
3231
3232                         /* Let's check if this linkEA still valid, before
3233                          * it might be packed into the RPC buffer. */
3234                         rc = mdd_lookup(env, &pobj->mod_obj, &lname,
3235                                         &info->mti_fid, NULL);
3236                         if (rc < 0 || !lu_fid_eq(&info->mti_fid,
3237                                                  mdd_object_fid(mdd_sobj)))
3238                                 GOTO(next_put, rc == -ENOENT ? 0 : rc);
3239
3240                         rc = __mdd_index_delete(env, pobj, tmp_name, 0, handle);
3241                         if (rc != 0)
3242                                 GOTO(next_put, rc);
3243
3244                         rc = __mdd_index_insert(env, pobj,
3245                                         mdd_object_fid(mdd_tobj),
3246                                         mdd_object_type(mdd_tobj),
3247                                         tmp_name, handle);
3248                         if (rc != 0)
3249                                 GOTO(next_put, rc);
3250
3251                         mdd_write_lock(env, mdd_tobj, MOR_SRC_CHILD);
3252                         rc = mdo_ref_add(env, mdd_tobj, handle);
3253                         mdd_write_unlock(env, mdd_tobj);
3254                         if (rc)
3255                                 GOTO(next_put, rc);
3256
3257                         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
3258                         mdo_ref_del(env, mdd_sobj, handle);
3259                         mdd_write_unlock(env, mdd_sobj);
3260                 }
3261 next_put:
3262                 mdd_object_put(env, pobj);
3263                 if (rc != 0)
3264                         break;
3265
3266                 ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
3267                                                          ldata->ld_reclen);
3268         }
3269
3270         RETURN(rc);
3271 }
3272
3273 static int mdd_migrate_xattrs(const struct lu_env *env,
3274                               struct mdd_object *mdd_sobj,
3275                               struct mdd_object *mdd_tobj)
3276 {
3277         struct mdd_thread_info  *info = mdd_env_info(env);
3278         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3279         char                    *xname;
3280         struct thandle          *handle;
3281         struct lu_buf           xbuf;
3282         int                     xlen;
3283         int                     rem;
3284         int                     xsize;
3285         int                     list_xsize;
3286         struct lu_buf           list_xbuf;
3287         int                     rc;
3288
3289         /* retrieve xattr list from the old object */
3290         list_xsize = mdo_xattr_list(env, mdd_sobj, &LU_BUF_NULL);
3291         if (list_xsize == -ENODATA)
3292                 return 0;
3293
3294         if (list_xsize < 0)
3295                 return list_xsize;
3296
3297         lu_buf_check_and_alloc(&info->mti_big_buf, list_xsize);
3298         if (info->mti_big_buf.lb_buf == NULL)
3299                 return -ENOMEM;
3300
3301         list_xbuf.lb_buf = info->mti_big_buf.lb_buf;
3302         list_xbuf.lb_len = list_xsize;
3303         rc = mdo_xattr_list(env, mdd_sobj, &list_xbuf);
3304         if (rc < 0)
3305                 return rc;
3306         rc = 0;
3307         rem = list_xsize;
3308         xname = list_xbuf.lb_buf;
3309         while (rem > 0) {
3310                 xlen = strnlen(xname, rem - 1) + 1;
3311                 if (strcmp(XATTR_NAME_LMA, xname) == 0 ||
3312                     strcmp(XATTR_NAME_LMV, xname) == 0)
3313                         goto next;
3314
3315                 /* For directory, if there are default layout, migrate here */
3316                 if (strcmp(XATTR_NAME_LOV, xname) == 0 &&
3317                     !S_ISDIR(lu_object_attr(&mdd_sobj->mod_obj.mo_lu)))
3318                         goto next;
3319
3320                 xsize = mdo_xattr_get(env, mdd_sobj, &LU_BUF_NULL, xname);
3321                 if (xsize == -ENODATA)
3322                         goto next;
3323                 if (xsize < 0)
3324                         GOTO(out, rc);
3325
3326                 lu_buf_check_and_alloc(&info->mti_link_buf, xsize);
3327                 if (info->mti_link_buf.lb_buf == NULL)
3328                         GOTO(out, rc = -ENOMEM);
3329
3330                 xbuf.lb_len = xsize;
3331                 xbuf.lb_buf = info->mti_link_buf.lb_buf;
3332                 rc = mdo_xattr_get(env, mdd_sobj, &xbuf, xname);
3333                 if (rc == -ENODATA)
3334                         goto next;
3335                 if (rc < 0)
3336                         GOTO(out, rc);
3337
3338                 handle = mdd_trans_create(env, mdd);
3339                 if (IS_ERR(handle))
3340                         GOTO(out, rc = PTR_ERR(handle));
3341
3342                 rc = mdo_declare_xattr_set(env, mdd_tobj, &xbuf, xname, 0,
3343                                            handle);
3344                 if (rc != 0)
3345                         GOTO(stop_trans, rc);
3346                 /* Note: this transaction is part of migration, and it is not
3347                  * the last step of migration, so we set th_local = 1 to avoid
3348                  * update last rcvd for this transaction */
3349                 handle->th_local = 1;
3350                 rc = mdd_trans_start(env, mdd, handle);
3351                 if (rc != 0)
3352                         GOTO(stop_trans, rc);
3353
3354 again:
3355                 rc = mdo_xattr_set(env, mdd_tobj, &xbuf, xname, 0, handle);
3356                 if (rc == -EEXIST)
3357                         GOTO(stop_trans, rc = 0);
3358
3359                 if (unlikely(rc == -ENOSPC &&
3360                              strcmp(xname, XATTR_NAME_LINK) == 0)) {
3361                         rc = linkea_overflow_shrink(
3362                                         (struct linkea_data *)(xbuf.lb_buf));
3363                         if (likely(rc > 0)) {
3364                                 xbuf.lb_len = rc;
3365                                 goto again;
3366                         }
3367                 }
3368
3369                 if (rc != 0)
3370                         GOTO(stop_trans, rc);
3371 stop_trans:
3372                 rc = mdd_trans_stop(env, mdd, rc, handle);
3373                 if (rc != 0)
3374                         GOTO(out, rc);
3375 next:
3376                 rem -= xlen;
3377                 memmove(xname, xname + xlen, rem);
3378         }
3379 out:
3380         return rc;
3381 }
3382
3383 static int mdd_declare_migrate_create(const struct lu_env *env,
3384                                       struct mdd_object *mdd_pobj,
3385                                       struct mdd_object *mdd_sobj,
3386                                       struct mdd_object *mdd_tobj,
3387                                       struct md_op_spec *spec,
3388                                       struct lu_attr *la,
3389                                       union lmv_mds_md *mgr_ea,
3390                                       struct linkea_data *ldata,
3391                                       struct thandle *handle)
3392 {
3393         struct lu_attr          *la_flag = MDD_ENV_VAR(env, la_for_fix);
3394         const struct lu_buf     *buf;
3395         int                     rc;
3396         int                     mgr_easize;
3397
3398         rc = mdd_declare_object_create_internal(env, mdd_pobj, mdd_tobj, la,
3399                                                 handle, spec, NULL);
3400         if (rc != 0)
3401                 return rc;
3402
3403         rc = mdd_declare_object_initialize(env, mdd_pobj, mdd_tobj, la,
3404                                            handle);
3405         if (rc != 0)
3406                 return rc;
3407
3408         if (S_ISLNK(la->la_mode)) {
3409                 const char *target_name = spec->u.sp_symname;
3410                 int sym_len = strlen(target_name);
3411                 const struct lu_buf *buf;
3412
3413                 buf = mdd_buf_get_const(env, target_name, sym_len);
3414                 rc = dt_declare_record_write(env, mdd_object_child(mdd_tobj),
3415                                              buf, 0, handle);
3416                 if (rc != 0)
3417                         return rc;
3418         } else if (S_ISDIR(la->la_mode) && ldata != NULL) {
3419                 rc = mdd_declare_links_add(env, mdd_tobj, handle, ldata);
3420                 if (rc != 0)
3421                         return rc;
3422         }
3423
3424         if (spec->u.sp_ea.eadata != NULL && spec->u.sp_ea.eadatalen != 0) {
3425                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
3426                                         spec->u.sp_ea.eadatalen);
3427                 rc = mdo_declare_xattr_set(env, mdd_tobj, buf, XATTR_NAME_LOV,
3428                                            0, handle);
3429                 if (rc)
3430                         return rc;
3431         }
3432
3433         mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
3434         buf = mdd_buf_get_const(env, mgr_ea, mgr_easize);
3435         rc = mdo_declare_xattr_set(env, mdd_sobj, buf, XATTR_NAME_LMV,
3436                                    0, handle);
3437         if (rc)
3438                 return rc;
3439
3440         la_flag->la_valid = LA_FLAGS;
3441         la_flag->la_flags = la->la_flags | LUSTRE_IMMUTABLE_FL;
3442         rc = mdo_declare_attr_set(env, mdd_sobj, la_flag, handle);
3443
3444         return rc;
3445 }
3446
3447 static int mdd_migrate_create(const struct lu_env *env,
3448                               struct mdd_object *mdd_pobj,
3449                               struct mdd_object *mdd_sobj,
3450                               struct mdd_object *mdd_tobj,
3451                               const struct lu_name *lname,
3452                               struct lu_attr *la)
3453 {
3454         struct mdd_thread_info  *info = mdd_env_info(env);
3455         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3456         struct md_op_spec       *spec = &info->mti_spec;
3457         struct lu_buf           lmm_buf = { NULL };
3458         struct lu_buf           link_buf = { NULL };
3459         const struct lu_buf     *buf;
3460         struct thandle          *handle;
3461         struct lmv_mds_md_v1    *mgr_ea;
3462         struct lu_attr          *la_flag = MDD_ENV_VAR(env, la_for_fix);
3463         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
3464         int                     mgr_easize;
3465         struct linkea_data      *ldata = &mdd_env_info(env)->mti_link_data;
3466         int                     rc;
3467         ENTRY;
3468
3469         /* prepare spec for create */
3470         memset(spec, 0, sizeof(*spec));
3471         spec->sp_cr_lookup = 0;
3472         spec->sp_feat = &dt_directory_features;
3473         if (S_ISLNK(la->la_mode)) {
3474                 buf = lu_buf_check_and_alloc(
3475                                 &mdd_env_info(env)->mti_big_buf,
3476                                 la->la_size + 1);
3477                 link_buf = *buf;
3478                 link_buf.lb_len = la->la_size + 1;
3479                 memset(link_buf.lb_buf, 0, link_buf.lb_len);
3480                 rc = mdd_readlink(env, &mdd_sobj->mod_obj, &link_buf);
3481                 if (rc <= 0) {
3482                         rc = rc != 0 ? rc : -EFAULT;
3483                         CERROR("%s: "DFID" readlink failed: rc = %d\n",
3484                                mdd2obd_dev(mdd)->obd_name,
3485                                PFID(mdd_object_fid(mdd_sobj)), rc);
3486                         RETURN(rc);
3487                 }
3488                 spec->u.sp_symname = link_buf.lb_buf;
3489         } else if (S_ISREG(la->la_mode)) {
3490                 /* retrieve lov of the old object */
3491                 rc = mdd_get_lov_ea(env, mdd_sobj, &lmm_buf);
3492                 if (rc != 0 && rc != -ENODATA)
3493                         RETURN(rc);
3494                 if (lmm_buf.lb_buf != NULL && lmm_buf.lb_len != 0) {
3495                         spec->u.sp_ea.eadata = lmm_buf.lb_buf;
3496                         spec->u.sp_ea.eadatalen = lmm_buf.lb_len;
3497                         spec->sp_cr_flags |= MDS_OPEN_HAS_EA;
3498                 }
3499         } else if (S_ISDIR(la->la_mode)) {
3500                 rc = mdd_links_read_with_rec(env, mdd_sobj, ldata);
3501                 if (rc == -ENODATA) {
3502                         /* ignore the non-linkEA error */
3503                         ldata = NULL;
3504                         rc = 0;
3505                 }
3506                 if (rc < 0)
3507                         RETURN(rc);
3508         }
3509
3510         mgr_ea = (struct lmv_mds_md_v1 *)info->mti_xattr_buf;
3511         memset(mgr_ea, 0, sizeof(*mgr_ea));
3512         mgr_ea->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
3513         mgr_ea->lmv_stripe_count = cpu_to_le32(2);
3514         mgr_ea->lmv_master_mdt_index = mdd_seq_site(mdd)->ss_node_id;
3515         mgr_ea->lmv_hash_type = cpu_to_le32(LMV_HASH_FLAG_MIGRATION);
3516         fid_cpu_to_le(&mgr_ea->lmv_stripe_fids[0], mdd_object_fid(mdd_sobj));
3517         fid_cpu_to_le(&mgr_ea->lmv_stripe_fids[1], mdd_object_fid(mdd_tobj));
3518
3519         mdd_object_make_hint(env, mdd_pobj, mdd_tobj, la, spec, hint);
3520
3521         handle = mdd_trans_create(env, mdd);
3522         if (IS_ERR(handle))
3523                 GOTO(out_free, rc = PTR_ERR(handle));
3524
3525         /* Note: this transaction is part of migration, and it is not
3526          * the last step of migration, so we set th_local = 1 to avoid
3527          * update last rcvd for this transaction */
3528         handle->th_local = 1;
3529         rc = mdd_declare_migrate_create(env, mdd_pobj, mdd_sobj, mdd_tobj,
3530                                         spec, la,
3531                                         (union lmv_mds_md *)info->mti_xattr_buf,
3532                                         ldata, handle);
3533         if (rc != 0)
3534                 GOTO(stop_trans, rc);
3535
3536         rc = mdd_trans_start(env, mdd, handle);
3537         if (rc != 0)
3538                 GOTO(stop_trans, rc);
3539
3540         /* don't set nlink from the original object */
3541         la->la_valid &= ~LA_NLINK;
3542
3543         /* create the target object */
3544         rc = mdd_object_create(env, mdd_pobj, mdd_tobj, la, spec, NULL, NULL,
3545                                hint, handle);
3546         if (rc != 0)
3547                 GOTO(stop_trans, rc);
3548
3549         if (S_ISDIR(la->la_mode) && ldata != NULL) {
3550                 rc = mdd_links_write(env, mdd_tobj, ldata, handle);
3551                 if (rc != 0)
3552                         GOTO(stop_trans, rc);
3553         }
3554
3555         /* Set MIGRATE EA on the source inode, so once the migration needs
3556          * to be re-done during failover, the re-do process can locate the
3557          * target object which is already being created. */
3558         mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
3559         buf = mdd_buf_get_const(env, mgr_ea, mgr_easize);
3560         rc = mdo_xattr_set(env, mdd_sobj, buf, XATTR_NAME_LMV, 0, handle);
3561         if (rc != 0)
3562                 GOTO(stop_trans, rc);
3563
3564         /* Set immutable flag, so any modification is disabled until
3565          * the migration is done. Once the migration is interrupted,
3566          * if the resume process find the migrating object has both
3567          * IMMUTALBE flag and MIGRATE EA, it need to clear IMMUTABLE
3568          * flag and approve the migration */
3569         la_flag->la_valid = LA_FLAGS;
3570         la_flag->la_flags = la->la_flags | LUSTRE_IMMUTABLE_FL;
3571         rc = mdo_attr_set(env, mdd_sobj, la_flag, handle);
3572 stop_trans:
3573         if (handle != NULL)
3574                 rc = mdd_trans_stop(env, mdd, rc, handle);
3575 out_free:
3576         if (lmm_buf.lb_buf != NULL)
3577                 OBD_FREE(lmm_buf.lb_buf, lmm_buf.lb_len);
3578         RETURN(rc);
3579 }
3580
3581 static int mdd_migrate_entries(const struct lu_env *env,
3582                                struct mdd_object *mdd_sobj,
3583                                struct mdd_object *mdd_tobj)
3584 {
3585         struct dt_object        *next = mdd_object_child(mdd_sobj);
3586         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3587         struct dt_object        *dt_tobj = mdd_object_child(mdd_tobj);
3588         struct thandle          *handle;
3589         struct dt_it            *it;
3590         const struct dt_it_ops  *iops;
3591         int                      result;
3592         struct lu_dirent        *ent;
3593         int                      rc;
3594         ENTRY;
3595
3596         OBD_ALLOC(ent, NAME_MAX + sizeof(*ent) + 1);
3597         if (ent == NULL)
3598                 RETURN(-ENOMEM);
3599
3600         if (!dt_try_as_dir(env, next))
3601                 GOTO(out_ent, rc = -ENOTDIR);
3602         /*
3603          * iterate directories
3604          */
3605         iops = &next->do_index_ops->dio_it;
3606         it = iops->init(env, next, LUDA_FID | LUDA_TYPE);
3607         if (IS_ERR(it))
3608                 GOTO(out_ent, rc = PTR_ERR(it));
3609
3610         rc = iops->load(env, it, 0);
3611         if (rc == 0)
3612                 rc = iops->next(env, it);
3613         else if (rc > 0)
3614                 rc = 0;
3615         /*
3616          * At this point and across for-loop:
3617          *
3618          *  rc == 0 -> ok, proceed.
3619          *  rc >  0 -> end of directory.
3620          *  rc <  0 -> error.
3621          */
3622         do {
3623                 struct mdd_object       *child;
3624                 char                    *name = mdd_env_info(env)->mti_key;
3625                 int                     len;
3626                 int                     is_dir;
3627                 bool                    target_exist = false;
3628
3629                 len = iops->key_size(env, it);
3630                 if (len == 0)
3631                         goto next;
3632
3633                 result = iops->rec(env, it, (struct dt_rec *)ent,
3634                                    LUDA_FID | LUDA_TYPE);
3635                 if (result == -ESTALE)
3636                         goto next;
3637                 if (result != 0) {
3638                         rc = result;
3639                         goto out;
3640                 }
3641
3642                 fid_le_to_cpu(&ent->lde_fid, &ent->lde_fid);
3643
3644                 /* Insert new fid with target name into target dir */
3645                 if ((ent->lde_namelen == 1 && ent->lde_name[0] == '.') ||
3646                     (ent->lde_namelen == 2 && ent->lde_name[0] == '.' &&
3647                      ent->lde_name[1] == '.'))
3648                         goto next;
3649
3650                 child = mdd_object_find(env, mdd, &ent->lde_fid);
3651                 if (IS_ERR(child))
3652                         GOTO(out, rc = PTR_ERR(child));
3653
3654                 mdd_write_lock(env, child, MOR_SRC_CHILD);
3655                 is_dir = S_ISDIR(mdd_object_type(child));
3656
3657                 snprintf(name, ent->lde_namelen + 1, "%s", ent->lde_name);
3658
3659                 /* Check whether the name has been inserted to the target */
3660                 if (dt_try_as_dir(env, dt_tobj)) {
3661                         struct lu_fid *fid = &mdd_env_info(env)->mti_fid2;
3662
3663                         rc = dt_lookup(env, dt_tobj, (struct dt_rec *)fid,
3664                                        (struct dt_key *)name);
3665                         if (unlikely(rc == 0))
3666                                 target_exist = true;
3667                 }
3668
3669                 handle = mdd_trans_create(env, mdd);
3670                 if (IS_ERR(handle))
3671                         GOTO(out_put, rc = PTR_ERR(handle));
3672
3673                 /* Note: this transaction is part of migration, and it is not
3674                  * the last step of migration, so we set th_local = 1 to avoid
3675                  * updating last rcvd for this transaction */
3676                 handle->th_local = 1;
3677                 if (likely(!target_exist)) {
3678                         rc = mdo_declare_index_insert(env, mdd_tobj,
3679                                                       &ent->lde_fid,
3680                                                       mdd_object_type(child),
3681                                                       name, handle);
3682                         if (rc != 0)
3683                                 GOTO(out_put, rc);
3684
3685                         if (is_dir) {
3686                                 rc = mdo_declare_ref_add(env, mdd_tobj, handle);
3687                                 if (rc != 0)
3688                                         GOTO(out_put, rc);
3689                         }
3690                 }
3691
3692                 rc = mdo_declare_index_delete(env, mdd_sobj, name, handle);
3693                 if (rc != 0)
3694                         GOTO(out_put, rc);
3695
3696                 if (is_dir) {
3697                         rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3698                         if (rc != 0)
3699                                 GOTO(out_put, rc);
3700
3701                         /* Update .. for child */
3702                         rc = mdo_declare_index_delete(env, child, dotdot,
3703                                                       handle);
3704                         if (rc != 0)
3705                                 GOTO(out_put, rc);
3706
3707                         rc = mdo_declare_index_insert(env, child,
3708                                                       mdd_object_fid(mdd_tobj),
3709                                                       S_IFDIR, dotdot, handle);
3710                         if (rc != 0)
3711                                 GOTO(out_put, rc);
3712                 }
3713
3714                 rc = mdd_linkea_declare_update_child(env, mdd_sobj,mdd_tobj,
3715                                                      child, name,
3716                                                      strlen(name),
3717                                                      handle);
3718                 if (rc != 0)
3719                         GOTO(out_put, rc);
3720
3721                 rc = mdd_trans_start(env, mdd, handle);
3722                 if (rc != 0) {
3723                         CERROR("%s: transaction start failed: rc = %d\n",
3724                                mdd2obd_dev(mdd)->obd_name, rc);
3725                         GOTO(out_put, rc);
3726                 }
3727
3728                 if (likely(!target_exist)) {
3729                         rc = __mdd_index_insert(env, mdd_tobj, &ent->lde_fid,
3730                                                 mdd_object_type(child),
3731                                                 name, handle);
3732                         if (rc != 0)
3733                                 GOTO(out_put, rc);
3734                 }
3735
3736                 rc = __mdd_index_delete(env, mdd_sobj, name, is_dir, handle);
3737                 if (rc != 0)
3738                         GOTO(out_put, rc);
3739
3740                 if (is_dir) {
3741                         rc = __mdd_index_delete_only(env, child, dotdot,
3742                                                      handle);
3743                         if (rc != 0)
3744                                 GOTO(out_put, rc);
3745
3746                         rc = __mdd_index_insert_only(env, child,
3747                                          mdd_object_fid(mdd_tobj), S_IFDIR,
3748                                          dotdot, handle);
3749                         if (rc != 0)
3750                                 GOTO(out_put, rc);
3751                 }
3752
3753                 rc = mdd_linkea_update_child(env, mdd_sobj, mdd_tobj,
3754                                              child, name,
3755                                              strlen(name), handle);
3756
3757 out_put:
3758                 mdd_write_unlock(env, child);
3759                 mdd_object_put(env, child);
3760                 rc = mdd_trans_stop(env, mdd, rc, handle);
3761                 if (rc != 0)
3762                         GOTO(out, rc);
3763 next:
3764                 result = iops->next(env, it);
3765                 if (OBD_FAIL_CHECK(OBD_FAIL_MIGRATE_ENTRIES))
3766                         GOTO(out, rc = -EINTR);
3767
3768                 if (result == -ESTALE)
3769                         goto next;
3770         } while (result == 0);
3771 out:
3772         iops->put(env, it);
3773         iops->fini(env, it);
3774 out_ent:
3775         OBD_FREE(ent, NAME_MAX + sizeof(*ent) + 1);
3776         RETURN(rc);
3777 }
3778
3779 static int mdd_declare_update_linkea(const struct lu_env *env,
3780                                      struct mdd_object *mdd_pobj,
3781                                      struct mdd_object *mdd_sobj,
3782                                      struct mdd_object *mdd_tobj,
3783                                      const struct lu_name *child_name,
3784                                      struct linkea_data *ldata,
3785                                      struct thandle *handle)
3786 {
3787         return mdd_update_linkea_internal(env, mdd_pobj, mdd_sobj, mdd_tobj,
3788                                           child_name, ldata, handle, 1);
3789 }
3790
3791 static int mdd_update_linkea(const struct lu_env *env,
3792                              struct mdd_object *mdd_pobj,
3793                              struct mdd_object *mdd_sobj,
3794                              struct mdd_object *mdd_tobj,
3795                              const struct lu_name *child_name,
3796                              struct linkea_data *ldata,
3797                              struct thandle *handle)
3798 {
3799         return mdd_update_linkea_internal(env, mdd_pobj, mdd_sobj, mdd_tobj,
3800                                           child_name, ldata, handle, 0);
3801 }
3802
3803 static int mdd_declare_migrate_update_name(const struct lu_env *env,
3804                                            struct mdd_object *mdd_pobj,
3805                                            struct mdd_object *mdd_sobj,
3806                                            struct mdd_object *mdd_tobj,
3807                                            const struct lu_name *lname,
3808                                            struct lu_attr *la,
3809                                            struct lu_attr *parent_la,
3810                                            struct linkea_data *ldata,
3811                                            struct thandle *handle)
3812 {
3813         struct mdd_device *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3814         struct lu_attr *la_flag = MDD_ENV_VAR(env, tattr);
3815         int rc;
3816
3817         /* Revert IMMUTABLE flag */
3818         la_flag->la_valid = LA_FLAGS;
3819         la_flag->la_flags = la->la_flags & ~LUSTRE_IMMUTABLE_FL;
3820         rc = mdo_declare_attr_set(env, mdd_sobj, la_flag, handle);
3821         if (rc != 0)
3822                 return rc;
3823
3824         /* delete entry from source dir */
3825         rc = mdo_declare_index_delete(env, mdd_pobj, lname->ln_name, handle);
3826         if (rc != 0)
3827                 return rc;
3828
3829         if (ldata->ld_buf != NULL) {
3830                 rc = mdd_declare_update_linkea(env, mdd_pobj, mdd_sobj,
3831                                                mdd_tobj, lname, ldata, handle);
3832                 if (rc != 0)
3833                         return rc;
3834         }
3835
3836         if (S_ISREG(mdd_object_type(mdd_sobj))) {
3837                 rc = mdo_declare_xattr_del(env, mdd_sobj, XATTR_NAME_LOV,
3838                                            handle);
3839                 if (rc != 0)
3840                         return rc;
3841
3842                 handle->th_complex = 1;
3843                 rc = mdo_declare_xattr_set(env, mdd_tobj, NULL,
3844                                            XATTR_NAME_FID,
3845                                            LU_XATTR_REPLACE, handle);
3846                 if (rc < 0)
3847                         return rc;
3848         }
3849
3850         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
3851                 rc = mdo_declare_ref_del(env, mdd_pobj, handle);
3852                 if (rc != 0)
3853                         return rc;
3854         }
3855
3856         /* new name */
3857         rc = mdo_declare_index_insert(env, mdd_pobj, mdo2fid(mdd_tobj),
3858                                       mdd_object_type(mdd_tobj),
3859                                       lname->ln_name, handle);
3860         if (rc != 0)
3861                 return rc;
3862
3863         rc = mdd_declare_links_add(env, mdd_tobj, handle, NULL);
3864         if (rc != 0)
3865                 return rc;
3866
3867         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
3868                 rc = mdo_declare_ref_add(env, mdd_pobj, handle);
3869                 if (rc != 0)
3870                         return rc;
3871         }
3872
3873         /* delete old object */
3874         rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3875         if (rc != 0)
3876                 return rc;
3877
3878         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
3879                 /* delete old object */
3880                 rc = mdo_declare_ref_del(env, mdd_sobj, handle);
3881                 if (rc != 0)
3882                         return rc;
3883                 /* set nlink to 0 */
3884                 rc = mdo_declare_attr_set(env, mdd_sobj, la, handle);
3885                 if (rc != 0)
3886                         return rc;
3887         }
3888
3889         rc = mdd_declare_finish_unlink(env, mdd_sobj, handle);
3890         if (rc)
3891                 return rc;
3892
3893         rc = mdo_declare_attr_set(env, mdd_pobj, parent_la, handle);
3894         if (rc != 0)
3895                 return rc;
3896
3897         rc = mdd_declare_changelog_store(env, mdd, lname, NULL, handle);
3898
3899         return rc;
3900 }
3901
3902 static int mdd_migrate_update_name(const struct lu_env *env,
3903                                    struct mdd_object *mdd_pobj,
3904                                    struct mdd_object *mdd_sobj,
3905                                    struct mdd_object *mdd_tobj,
3906                                    const struct lu_name *lname,
3907                                    struct md_attr *ma)
3908 {
3909         struct lu_attr          *p_la = MDD_ENV_VAR(env, la_for_fix);
3910         struct lu_attr          *so_attr = MDD_ENV_VAR(env, cattr);
3911         struct lu_attr          *la_flag = MDD_ENV_VAR(env, tattr);
3912         struct mdd_device       *mdd = mdo2mdd(&mdd_sobj->mod_obj);
3913         struct linkea_data      *ldata = &mdd_env_info(env)->mti_link_data;
3914         struct thandle          *handle;
3915         int                     is_dir = S_ISDIR(mdd_object_type(mdd_sobj));
3916         const char              *name = lname->ln_name;
3917         int                     rc;
3918         ENTRY;
3919
3920         /* update time for parent */
3921         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
3922         p_la->la_ctime = p_la->la_mtime = ma->ma_attr.la_ctime;
3923         p_la->la_valid = LA_CTIME;
3924
3925         rc = mdd_la_get(env, mdd_sobj, so_attr);
3926         if (rc != 0)
3927                 RETURN(rc);
3928
3929         ldata->ld_buf = NULL;
3930         rc = mdd_links_read(env, mdd_sobj, ldata);
3931         if (rc != 0 && rc != -ENOENT && rc != -ENODATA)
3932                 RETURN(rc);
3933
3934         handle = mdd_trans_create(env, mdd);
3935         if (IS_ERR(handle))
3936                 RETURN(PTR_ERR(handle));
3937
3938         rc = mdd_declare_migrate_update_name(env, mdd_pobj, mdd_sobj, mdd_tobj,
3939                                              lname, so_attr, p_la, ldata,
3940                                              handle);
3941         if (rc != 0) {
3942                 /* If the migration can not be fit in one transaction, just
3943                  * leave it in the original MDT */
3944                 if (rc == -E2BIG)
3945                         GOTO(stop_trans, rc = 0);
3946                 else
3947                         GOTO(stop_trans, rc);
3948         }
3949
3950         CDEBUG(D_INFO, "%s: update "DFID"/"DFID" with %s:"DFID"\n",
3951                mdd2obd_dev(mdd)->obd_name, PFID(mdd_object_fid(mdd_pobj)),
3952                PFID(mdd_object_fid(mdd_sobj)), lname->ln_name,
3953                PFID(mdd_object_fid(mdd_tobj)));
3954
3955         rc = mdd_trans_start(env, mdd, handle);
3956         if (rc != 0)
3957                 GOTO(stop_trans, rc);
3958
3959         /* Revert IMMUTABLE flag */
3960         la_flag->la_valid = LA_FLAGS;
3961         la_flag->la_flags = so_attr->la_flags & ~LUSTRE_IMMUTABLE_FL;
3962         rc = mdo_attr_set(env, mdd_sobj, la_flag, handle);
3963         if (rc != 0)
3964                 GOTO(stop_trans, rc);
3965
3966         /* Remove source name from source directory */
3967         rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle);
3968         if (rc != 0)
3969                 GOTO(stop_trans, rc);
3970
3971         if (ldata->ld_buf != NULL) {
3972                 rc = mdd_update_linkea(env, mdd_pobj, mdd_sobj, mdd_tobj,
3973                                        lname, ldata, handle);
3974                 if (rc != 0)
3975                         GOTO(stop_trans, rc);
3976
3977                 /*  linkea update might decrease the source object
3978                  *  nlink, let's get the attr again after ref_del */
3979                 rc = mdd_la_get(env, mdd_sobj, so_attr);
3980                 if (rc != 0)
3981                         GOTO(stop_trans, rc);
3982         }
3983
3984         if (S_ISREG(so_attr->la_mode)) {
3985                 if (so_attr->la_nlink == 1) {
3986                         rc = mdo_xattr_del(env, mdd_sobj, XATTR_NAME_LOV,
3987                                            handle);
3988                         if (rc != 0 && rc != -ENODATA)
3989                                 GOTO(stop_trans, rc);
3990
3991                         rc = mdo_xattr_set(env, mdd_tobj, NULL,
3992                                            XATTR_NAME_FID,
3993                                            LU_XATTR_REPLACE, handle);
3994                         if (rc < 0)
3995                                 GOTO(stop_trans, rc);
3996                 }
3997         }
3998
3999         /* Insert new fid with target name into target dir */
4000         rc = __mdd_index_insert(env, mdd_pobj, mdd_object_fid(mdd_tobj),
4001                                 mdd_object_type(mdd_tobj), name, handle);
4002         if (rc != 0)
4003                 GOTO(stop_trans, rc);
4004
4005         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
4006
4007         mdd_sobj->mod_flags |= DEAD_OBJ;
4008         rc = mdd_mark_orphan_object(env, mdd_sobj, handle, false);
4009         if (rc != 0)
4010                 GOTO(out_unlock, rc);
4011
4012         rc = __mdd_orphan_add(env, mdd_sobj, handle);
4013         if (rc != 0)
4014                 GOTO(out_unlock, rc);
4015
4016         mdo_ref_del(env, mdd_sobj, handle);
4017         if (is_dir)
4018                 mdo_ref_del(env, mdd_sobj, handle);
4019
4020         /* Get the attr again after ref_del */
4021         rc = mdd_la_get(env, mdd_sobj, so_attr);
4022         if (rc != 0)
4023                 GOTO(out_unlock, rc);
4024
4025         ma->ma_attr = *so_attr;
4026         ma->ma_valid |= MA_INODE;
4027
4028         rc = mdd_attr_set_internal(env, mdd_pobj, p_la, handle, 0);
4029         if (rc != 0)
4030                 GOTO(out_unlock, rc);
4031
4032         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0, mdd_tobj,
4033                                mdo2fid(mdd_pobj), mdo2fid(mdd_sobj),
4034                                mdo2fid(mdd_pobj), lname, lname, handle);
4035         if (rc != 0) {
4036                 CWARN("%s: changelog for migrate %s "DFID
4037                       "under "DFID" failed: rc = %d\n",
4038                       mdd2obd_dev(mdd)->obd_name, lname->ln_name,
4039                       PFID(mdd_object_fid(mdd_sobj)),
4040                       PFID(mdd_object_fid(mdd_pobj)), rc);
4041                 /* Sigh, there are no easy way to migrate back the object, so
4042                  * let's reset the result to 0 for now XXX */
4043                 rc = 0;
4044         }
4045 out_unlock:
4046         mdd_write_unlock(env, mdd_sobj);
4047
4048 stop_trans:
4049         rc = mdd_trans_stop(env, mdd, rc, handle);
4050
4051         RETURN(rc);
4052 }
4053
4054 static int mdd_fld_lookup(const struct lu_env *env, struct mdd_device *mdd,
4055                           const struct lu_fid *fid, __u32 *mdt_index)
4056 {
4057         struct lu_seq_range *range = &mdd_env_info(env)->mti_range;
4058         struct seq_server_site *ss;
4059         int rc;
4060
4061         ss = mdd->mdd_md_dev.md_lu_dev.ld_site->ld_seq_site;
4062
4063         range->lsr_flags = LU_SEQ_RANGE_MDT;
4064         rc = fld_server_lookup(env, ss->ss_server_fld, fid->f_seq, range);
4065         if (rc != 0)
4066                 return rc;
4067
4068         *mdt_index = range->lsr_index;
4069
4070         return 0;
4071 }
4072 /**
4073  * Check whether we should migrate the file/dir
4074  * return val
4075  *      < 0  permission check failed or other error.
4076  *      = 0  the file can be migrated.
4077  *      > 0  the file does not need to be migrated, mostly
4078  *           for multiple link file
4079  **/
4080 static int mdd_migrate_sanity_check(const struct lu_env *env,
4081                                     struct mdd_object *pobj,
4082                                     const struct lu_attr *pattr,
4083                                     struct mdd_object *sobj,
4084                                     struct lu_attr *sattr)
4085 {
4086         struct mdd_thread_info  *info = mdd_env_info(env);
4087         struct linkea_data      *ldata = &info->mti_link_data;
4088         struct mdd_device       *mdd = mdo2mdd(&pobj->mod_obj);
4089         int                     mgr_easize;
4090         struct lu_buf           *mgr_buf;
4091         int                     count;
4092         int                     rc;
4093         __u64 mdt_index;
4094         ENTRY;
4095
4096         mgr_easize = lmv_mds_md_size(2, LMV_MAGIC_V1);
4097         mgr_buf = lu_buf_check_and_alloc(&info->mti_big_buf, mgr_easize);
4098         if (mgr_buf->lb_buf == NULL)
4099                 RETURN(-ENOMEM);
4100
4101         rc = mdo_xattr_get(env, sobj, mgr_buf, XATTR_NAME_LMV);
4102         if (rc > 0) {
4103                 union lmv_mds_md *lmm = mgr_buf->lb_buf;
4104
4105                 /* If the object has migrateEA, it means IMMUTE flag
4106                  * is being set by previous migration process, so it
4107                  * needs to override the IMMUTE flag, otherwise the
4108                  * following sanity check will fail */
4109                 if (le32_to_cpu(lmm->lmv_md_v1.lmv_hash_type) &
4110                                                 LMV_HASH_FLAG_MIGRATION) {
4111                         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
4112
4113                         sattr->la_flags &= ~LUSTRE_IMMUTABLE_FL;
4114                         CDEBUG(D_HA, "%s: "DFID" override IMMUTE FLAG\n",
4115                                mdd2obd_dev(mdd)->obd_name,
4116                                PFID(mdd_object_fid(sobj)));
4117                 }
4118         }
4119
4120         rc = mdd_rename_sanity_check(env, pobj, pattr, pobj, pattr,
4121                                      sobj, sattr, NULL, NULL);
4122         if (rc != 0)
4123                 RETURN(rc);
4124
4125         /* Then it will check if the file should be migrated. If the file
4126          * has mulitple links, we only need migrate the file if all of its
4127          * entries has been migrated to the remote MDT */
4128         if (!S_ISREG(sattr->la_mode) || sattr->la_nlink < 2)
4129                 RETURN(0);
4130
4131         rc = mdd_links_read(env, sobj, ldata);
4132         if (rc != 0) {
4133                 /* For multiple links files, if there are no linkEA data at all,
4134                  * means the file might be created before linkEA is enabled, and
4135                  * all of its links should not be migrated yet, otherwise it
4136                  * should have some linkEA there */
4137                 if (rc == -ENOENT || rc == -ENODATA)
4138                         RETURN(1);
4139                 RETURN(rc);
4140         }
4141
4142         mdt_index = mdd->mdd_md_dev.md_lu_dev.ld_site->ld_seq_site->ss_node_id;
4143         /* If there are still links locally, then the file will not be
4144          * migrated. */
4145         LASSERT(ldata->ld_leh != NULL);
4146
4147         /* If the linkEA is overflow, then means there are some unknown name
4148          * entries under unknown parents, that will prevent the migration. */
4149         if (unlikely(ldata->ld_leh->leh_overflow_time))
4150                 RETURN(1);
4151
4152         ldata->ld_lee = (struct link_ea_entry *)(ldata->ld_leh + 1);
4153         for (count = 0; count < ldata->ld_leh->leh_reccount; count++) {
4154                 struct lu_name          lname;
4155                 struct lu_fid           fid;
4156                 __u32                   parent_mdt_index;
4157
4158                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
4159                                     &lname, &fid);
4160                 ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
4161                                                          ldata->ld_reclen);
4162
4163                 rc = mdd_fld_lookup(env, mdd, &fid, &parent_mdt_index);
4164                 if (rc != 0)
4165                         RETURN(rc);
4166
4167                 /* Migrate the object only if none of its parents are on the
4168                  * current MDT. */
4169                 if (parent_mdt_index != mdt_index)
4170                         continue;
4171
4172                 CDEBUG(D_INFO, DFID"still has local entry %.*s "DFID"\n",
4173                        PFID(mdd_object_fid(sobj)), lname.ln_namelen,
4174                        lname.ln_name, PFID(&fid));
4175                 rc = 1;
4176                 break;
4177         }
4178
4179         RETURN(rc);
4180 }
4181
4182 static int mdd_migrate(const struct lu_env *env, struct md_object *pobj,
4183                        struct md_object *sobj, const struct lu_name *lname,
4184                        struct md_object *tobj, struct md_attr *ma)
4185 {
4186         struct mdd_object       *mdd_pobj = md2mdd_obj(pobj);
4187         struct mdd_device       *mdd = mdo2mdd(pobj);
4188         struct mdd_object       *mdd_sobj = md2mdd_obj(sobj);
4189         struct mdd_object       *mdd_tobj = md2mdd_obj(tobj);
4190         struct lu_attr          *so_attr = MDD_ENV_VAR(env, cattr);
4191         struct lu_attr          *pattr = MDD_ENV_VAR(env, pattr);
4192         bool                    created = false;
4193         int                     rc;
4194
4195         ENTRY;
4196         /* If the file will being migrated, it will check whether
4197          * the file is being opened by someone else right now */
4198         mdd_read_lock(env, mdd_sobj, MOR_SRC_CHILD);
4199         if (mdd_sobj->mod_count > 0) {
4200                 CDEBUG(D_OTHER,
4201                        "%s: "DFID"%s is already opened count %d: rc = %d\n",
4202                        mdd2obd_dev(mdd)->obd_name,
4203                        PFID(mdd_object_fid(mdd_sobj)), lname->ln_name,
4204                        mdd_sobj->mod_count, -EBUSY);
4205                 mdd_read_unlock(env, mdd_sobj);
4206                 GOTO(put, rc = -EBUSY);
4207         }
4208         mdd_read_unlock(env, mdd_sobj);
4209
4210         rc = mdd_la_get(env, mdd_sobj, so_attr);
4211         if (rc != 0)
4212                 GOTO(put, rc);
4213
4214         rc = mdd_la_get(env, mdd_pobj, pattr);
4215         if (rc != 0)
4216                 GOTO(put, rc);
4217
4218         rc = mdd_migrate_sanity_check(env, mdd_pobj, pattr, mdd_sobj, so_attr);
4219         if (rc != 0) {
4220                 if (rc > 0)
4221                         rc = 0;
4222                 GOTO(put, rc);
4223         }
4224
4225         /* Sigh, it is impossible to finish all of migration in a single
4226          * transaction, for example migrating big directory entries to the
4227          * new MDT, it needs insert all of name entries of children in the
4228          * new directory.
4229          *
4230          * So migration will be done in multiple steps and transactions.
4231          *
4232          * 1. create an orphan object on the remote MDT in one transaction.
4233          * 2. migrate extend attributes to the new target file/directory.
4234          * 3. For directory, migrate the entries to the new MDT and update
4235          * linkEA of each children. Because we can not migrate all entries
4236          * in a single transaction, so the migrating directory will become
4237          * a striped directory during migration, so once the process is
4238          * interrupted, the directory is still accessible. (During lookup,
4239          * client will locate the name by searching both original and target
4240          * object).
4241          * 4. Finally, update the name/FID to point to the new file/directory
4242          * in a separate transaction.
4243          */
4244
4245         /* step 1: Check whether the orphan object has been created, and create
4246          * orphan object on the remote MDT if needed */
4247         if (!mdd_object_exists(mdd_tobj)) {
4248                 rc = mdd_migrate_create(env, mdd_pobj, mdd_sobj, mdd_tobj,
4249                                         lname, so_attr);
4250                 if (rc != 0)
4251                         GOTO(put, rc);
4252                 created = true;
4253         }
4254
4255         LASSERT(mdd_object_exists(mdd_tobj));
4256         /* step 2: migrate xattr */
4257         rc = mdd_migrate_xattrs(env, mdd_sobj, mdd_tobj);
4258         if (rc != 0)
4259                 GOTO(put, rc);
4260
4261         /* step 3: migrate name entries to the orphan object */
4262         if (S_ISDIR(lu_object_attr(&mdd_sobj->mod_obj.mo_lu))) {
4263                 rc = mdd_migrate_entries(env, mdd_sobj, mdd_tobj);
4264                 if (rc != 0)
4265                         GOTO(put, rc);
4266                 if (unlikely(OBD_FAIL_CHECK_RESET(OBD_FAIL_MIGRATE_NET_REP,
4267                                                   OBD_FAIL_MDS_REINT_NET_REP)))
4268                         GOTO(put, rc = 0);
4269         } else {
4270                 OBD_FAIL_TIMEOUT(OBD_FAIL_MIGRATE_DELAY, cfs_fail_val);
4271         }
4272
4273         LASSERT(mdd_object_exists(mdd_tobj));
4274         /* step 4: update name entry to the new object */
4275         rc = mdd_migrate_update_name(env, mdd_pobj, mdd_sobj, mdd_tobj, lname,
4276                                      ma);
4277         if (rc != 0)
4278                 GOTO(put, rc);
4279
4280         /* newly created target was not locked, don't cache its attributes */
4281         if (created)
4282                 mdd_invalidate(env, tobj);
4283 put:
4284         RETURN(rc);
4285 }
4286
4287 const struct md_dir_operations mdd_dir_ops = {
4288         .mdo_is_subdir     = mdd_is_subdir,
4289         .mdo_lookup        = mdd_lookup,
4290         .mdo_create        = mdd_create,
4291         .mdo_rename        = mdd_rename,
4292         .mdo_link          = mdd_link,
4293         .mdo_unlink        = mdd_unlink,
4294         .mdo_create_data   = mdd_create_data,
4295         .mdo_migrate       = mdd_migrate,
4296 };