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