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         attr->la_valid = LA_FLAGS;
1460         attr->la_flags = LUSTRE_ORPHAN_FL;
1461
1462         if (declare)
1463                 rc = mdo_declare_attr_set(env, obj, attr, handle);
1464         else
1465                 rc = mdo_attr_set(env, obj, attr, handle);
1466
1467         return rc;
1468 }
1469
1470 static int mdd_declare_finish_unlink(const struct lu_env *env,
1471                                      struct mdd_object *obj,
1472                                      struct thandle *handle)
1473 {
1474         int rc;
1475
1476         /* Sigh, we do not know if the unlink object will become orphan in
1477          * declare phase, but fortunately the flags here does not matter
1478          * in current declare implementation */
1479         rc = mdd_mark_orphan_object(env, obj, handle, true);
1480         if (rc != 0)
1481                 return rc;
1482
1483         rc = mdo_declare_destroy(env, obj, handle);
1484         if (rc != 0)
1485                 return rc;
1486
1487         rc = mdd_orphan_declare_insert(env, obj, mdd_object_type(obj), handle);
1488         if (rc != 0)
1489                 return rc;
1490
1491         return mdd_declare_links_del(env, obj, handle);
1492 }
1493
1494 /* caller should take a lock before calling */
1495 int mdd_finish_unlink(const struct lu_env *env,
1496                       struct mdd_object *obj, struct md_attr *ma,
1497                       const struct mdd_object *pobj,
1498                       const struct lu_name *lname,
1499                       struct thandle *th)
1500 {
1501         int rc = 0;
1502         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1503         ENTRY;
1504
1505         LASSERT(mdd_write_locked(env, obj) != 0);
1506
1507         if (ma->ma_attr.la_nlink == 0 || is_dir) {
1508                 /* add new orphan and the object
1509                  * will be deleted during mdd_close() */
1510                 obj->mod_flags |= DEAD_OBJ;
1511                 if (obj->mod_count) {
1512                         rc = mdd_orphan_insert(env, obj, th);
1513                         if (rc == 0)
1514                                 CDEBUG(D_HA, "Object "DFID" is inserted into "
1515                                         "orphan list, open count = %d\n",
1516                                         PFID(mdd_object_fid(obj)),
1517                                         obj->mod_count);
1518                         else
1519                                 CERROR("Object "DFID" fail to be an orphan, "
1520                                        "open count = %d, maybe cause failed "
1521                                        "open replay\n",
1522                                         PFID(mdd_object_fid(obj)),
1523                                         obj->mod_count);
1524
1525                         /* mark object as an orphan here, not
1526                          * before mdd_orphan_insert() as racing
1527                          * mdd_la_get() may propagate ORPHAN_OBJ
1528                          * causing the asserition */
1529                         rc = mdd_mark_orphan_object(env, obj, th, false);
1530                 } else {
1531                         rc = mdo_destroy(env, obj, th);
1532                 }
1533         } else if (!is_dir) {
1534                 /* old files may not have link ea; ignore errors */
1535                 mdd_links_del(env, obj, mdo2fid(pobj), lname, th);
1536         }
1537
1538         RETURN(rc);
1539 }
1540
1541 /*
1542  * pobj maybe NULL
1543  * has mdd_write_lock on cobj already, but not on pobj yet
1544  */
1545 int mdd_unlink_sanity_check(const struct lu_env *env, struct mdd_object *pobj,
1546                             const struct lu_attr *pattr,
1547                             struct mdd_object *cobj,
1548                             const struct lu_attr *cattr)
1549 {
1550         int rc;
1551         ENTRY;
1552
1553         rc = mdd_may_delete(env, pobj, pattr, cobj, cattr, NULL, 1, 1);
1554
1555         RETURN(rc);
1556 }
1557
1558 static int mdd_declare_unlink(const struct lu_env *env, struct mdd_device *mdd,
1559                               struct mdd_object *p, struct mdd_object *c,
1560                               const struct lu_name *name, struct md_attr *ma,
1561                               struct thandle *handle, int no_name, int is_dir)
1562 {
1563         struct lu_attr  *la = &mdd_env_info(env)->mti_la_for_fix;
1564         int              rc;
1565
1566         if (!OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1567                 if (likely(no_name == 0)) {
1568                         rc = mdo_declare_index_delete(env, p, name->ln_name,
1569                                                       handle);
1570                         if (rc != 0)
1571                                 return rc;
1572                 }
1573
1574                 if (is_dir != 0) {
1575                         rc = mdo_declare_ref_del(env, p, handle);
1576                         if (rc != 0)
1577                                 return rc;
1578                 }
1579         }
1580
1581         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1582         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1583         la->la_valid = LA_CTIME | LA_MTIME;
1584         rc = mdo_declare_attr_set(env, p, la, handle);
1585         if (rc)
1586                 return rc;
1587
1588         if (c != NULL) {
1589                 rc = mdo_declare_ref_del(env, c, handle);
1590                 if (rc)
1591                         return rc;
1592
1593                 rc = mdo_declare_ref_del(env, c, handle);
1594                 if (rc)
1595                         return rc;
1596
1597                 la->la_valid = LA_CTIME;
1598                 rc = mdo_declare_attr_set(env, c, la, handle);
1599                 if (rc)
1600                         return rc;
1601
1602                 rc = mdd_declare_finish_unlink(env, c, handle);
1603                 if (rc)
1604                         return rc;
1605
1606                 /* FIXME: need changelog for remove entry */
1607                 rc = mdd_declare_changelog_store(env, mdd, CL_UNLINK, name,
1608                                                  NULL, handle);
1609         }
1610
1611         return rc;
1612 }
1613
1614 /*
1615  * test if a file has an HSM archive
1616  * if HSM attributes are not found in ma update them from
1617  * HSM xattr
1618  */
1619 static bool mdd_hsm_archive_exists(const struct lu_env *env,
1620                                    struct mdd_object *obj,
1621                                    struct md_attr *ma)
1622 {
1623         ENTRY;
1624
1625         if (!(ma->ma_valid & MA_HSM)) {
1626                 /* no HSM MD provided, read xattr */
1627                 struct lu_buf   *hsm_buf;
1628                 const size_t     buflen = sizeof(struct hsm_attrs);
1629                 int              rc;
1630
1631                 hsm_buf = mdd_buf_get(env, NULL, 0);
1632                 lu_buf_alloc(hsm_buf, buflen);
1633                 rc = mdo_xattr_get(env, obj, hsm_buf, XATTR_NAME_HSM);
1634                 rc = lustre_buf2hsm(hsm_buf->lb_buf, rc, &ma->ma_hsm);
1635                 lu_buf_free(hsm_buf);
1636                 if (rc < 0)
1637                         RETURN(false);
1638
1639                 ma->ma_valid |= MA_HSM;
1640         }
1641         if (ma->ma_hsm.mh_flags & HS_EXISTS)
1642                 RETURN(true);
1643         RETURN(false);
1644 }
1645
1646 /**
1647  * Delete name entry and the object.
1648  * Note: no_name == 1 means it only destory the object, i.e. name_entry
1649  * does not exist for this object, and it could only happen during resending
1650  * of remote unlink. see the comments in mdt_reint_unlink. Unfortunately, lname
1651  * is also needed in this case(needed by changelog), so we have to add another
1652  * parameter(no_name)here. XXX: this is only needed in DNE phase I, on Phase II,
1653  * the ENOENT failure should be able to be fixed by redo mechanism.
1654  */
1655 static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
1656                       struct md_object *cobj, const struct lu_name *lname,
1657                       struct md_attr *ma, int no_name)
1658 {
1659         const char *name = lname->ln_name;
1660         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
1661         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
1662         struct lu_attr *la = &mdd_env_info(env)->mti_la_for_fix;
1663         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1664         struct mdd_object *mdd_cobj = NULL;
1665         struct mdd_device *mdd = mdo2mdd(pobj);
1666         struct thandle    *handle;
1667         int rc, is_dir = 0, cl_flags = 0;
1668         ENTRY;
1669
1670         /* let shutdown to start */
1671         CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLY_DATA_RACE, 1);
1672
1673         /* cobj == NULL means only delete name entry */
1674         if (likely(cobj != NULL)) {
1675                 mdd_cobj = md2mdd_obj(cobj);
1676                 if (mdd_object_exists(mdd_cobj) == 0)
1677                         RETURN(-ENOENT);
1678         }
1679
1680         rc = mdd_la_get(env, mdd_pobj, pattr);
1681         if (rc)
1682                 RETURN(rc);
1683
1684         if (likely(mdd_cobj != NULL)) {
1685                 /* fetch cattr */
1686                 rc = mdd_la_get(env, mdd_cobj, cattr);
1687                 if (rc)
1688                         RETURN(rc);
1689
1690                 is_dir = S_ISDIR(cattr->la_mode);
1691                 /* search for an existing archive.
1692                  * we should check ahead as the object
1693                  * can be destroyed in this transaction */
1694                 if (mdd_hsm_archive_exists(env, mdd_cobj, ma))
1695                         cl_flags |= CLF_UNLINK_HSM_EXISTS;
1696         }
1697
1698         rc = mdd_unlink_sanity_check(env, mdd_pobj, pattr, mdd_cobj, cattr);
1699         if (rc)
1700                 RETURN(rc);
1701
1702         handle = mdd_trans_create(env, mdd);
1703         if (IS_ERR(handle))
1704                 RETURN(PTR_ERR(handle));
1705
1706         rc = mdd_declare_unlink(env, mdd, mdd_pobj, mdd_cobj,
1707                                 lname, ma, handle, no_name, is_dir);
1708         if (rc)
1709                 GOTO(stop, rc);
1710
1711         rc = mdd_trans_start(env, mdd, handle);
1712         if (rc)
1713                 GOTO(stop, rc);
1714
1715         if (likely(mdd_cobj != NULL))
1716                 mdd_write_lock(env, mdd_cobj, DT_TGT_CHILD);
1717
1718         if (likely(no_name == 0) && !OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DANGLING2)) {
1719                 rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle);
1720                 if (rc)
1721                         GOTO(cleanup, rc);
1722         }
1723
1724         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_MUL_REF) ||
1725             OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_NAMEENTRY))
1726                 GOTO(cleanup, rc = 0);
1727
1728         if (likely(mdd_cobj != NULL)) {
1729                 rc = mdo_ref_del(env, mdd_cobj, handle);
1730                 if (rc != 0) {
1731                         __mdd_index_insert_only(env, mdd_pobj,
1732                                                 mdo2fid(mdd_cobj),
1733                                                 mdd_object_type(mdd_cobj),
1734                                                 name, handle);
1735                         GOTO(cleanup, rc);
1736                 }
1737
1738                 if (is_dir)
1739                         /* unlink dot */
1740                         mdo_ref_del(env, mdd_cobj, handle);
1741
1742                 /* fetch updated nlink */
1743                 rc = mdd_la_get(env, mdd_cobj, cattr);
1744                 if (rc)
1745                         GOTO(cleanup, rc);
1746         }
1747
1748         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1749         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1750
1751         la->la_valid = LA_CTIME | LA_MTIME;
1752         rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
1753         if (rc)
1754                 GOTO(cleanup, rc);
1755
1756         /* Enough for only unlink the entry */
1757         if (unlikely(mdd_cobj == NULL))
1758                 GOTO(stop, rc);
1759
1760         if (cattr->la_nlink > 0 || mdd_cobj->mod_count > 0) {
1761                 /* update ctime of an unlinked file only if it is still
1762                  * opened or a link still exists */
1763                 la->la_valid = LA_CTIME;
1764                 rc = mdd_update_time(env, mdd_cobj, cattr, la, handle);
1765                 if (rc)
1766                         GOTO(cleanup, rc);
1767         }
1768
1769         /* XXX: this transfer to ma will be removed with LOD/OSP */
1770         ma->ma_attr = *cattr;
1771         ma->ma_valid |= MA_INODE;
1772         rc = mdd_finish_unlink(env, mdd_cobj, ma, mdd_pobj, lname, handle);
1773         if (rc != 0)
1774                 GOTO(cleanup, rc);
1775
1776         /* fetch updated nlink */
1777         rc = mdd_la_get(env, mdd_cobj, cattr);
1778         /* if object is removed then we can't get its attrs,
1779          * use last get */
1780         if (rc == -ENOENT) {
1781                 cattr->la_nlink = 0;
1782                 rc = 0;
1783         }
1784
1785         if (cattr->la_nlink == 0) {
1786                 ma->ma_attr = *cattr;
1787                 ma->ma_valid |= MA_INODE;
1788         }
1789
1790         EXIT;
1791 cleanup:
1792         if (likely(mdd_cobj != NULL))
1793                 mdd_write_unlock(env, mdd_cobj);
1794
1795         if (rc == 0) {
1796                 if (cattr->la_nlink == 0)
1797                         cl_flags |= CLF_UNLINK_LAST;
1798                 else
1799                         cl_flags &= ~CLF_UNLINK_HSM_EXISTS;
1800
1801                 rc = mdd_changelog_ns_store(env, mdd,
1802                         is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,
1803                         mdd_cobj, mdo2fid(mdd_pobj), NULL, NULL, lname, NULL,
1804                         handle);
1805         }
1806
1807 stop:
1808         rc = mdd_trans_stop(env, mdd, rc, handle);
1809
1810         return rc;
1811 }
1812
1813 /*
1814  * The permission has been checked when obj created, no need check again.
1815  */
1816 static int mdd_cd_sanity_check(const struct lu_env *env,
1817                                struct mdd_object *obj)
1818 {
1819         ENTRY;
1820
1821         /* EEXIST check */
1822         if (!obj || mdd_is_dead_obj(obj))
1823                 RETURN(-ENOENT);
1824
1825         RETURN(0);
1826 }
1827
1828 static int mdd_create_data(const struct lu_env *env,
1829                            struct md_object *pobj,
1830                            struct md_object *cobj,
1831                            const struct md_op_spec *spec,
1832                            struct md_attr *ma)
1833 {
1834         struct mdd_device *mdd = mdo2mdd(cobj);
1835         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1836         struct mdd_object *son = md2mdd_obj(cobj);
1837         struct thandle    *handle;
1838         const struct lu_buf *buf;
1839         struct lu_attr    *attr = MDD_ENV_VAR(env, cattr);
1840         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
1841         int                rc;
1842         ENTRY;
1843
1844         rc = mdd_cd_sanity_check(env, son);
1845         if (rc)
1846                 RETURN(rc);
1847
1848         if (!md_should_create(spec->sp_cr_flags))
1849                 RETURN(0);
1850
1851         /*
1852          * there are following use cases for this function:
1853          * 1) late striping - file was created with MDS_OPEN_DELAY_CREATE
1854          *    striping can be specified or not
1855          * 2) CMD?
1856          */
1857         rc = mdd_la_get(env, son, attr);
1858         if (rc)
1859                 RETURN(rc);
1860
1861         /* calling ->ah_make_hint() is used to transfer information from parent */
1862         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
1863
1864         handle = mdd_trans_create(env, mdd);
1865         if (IS_ERR(handle))
1866                 GOTO(out_free, rc = PTR_ERR(handle));
1867
1868         /*
1869          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1870          * Should this be fixed?
1871          */
1872         CDEBUG(D_OTHER, "ea %p/%u, cr_flags %#llo, no_create %u\n",
1873                spec->u.sp_ea.eadata, spec->u.sp_ea.eadatalen,
1874                spec->sp_cr_flags, spec->no_create);
1875
1876         if (spec->no_create || (spec->sp_cr_flags & MDS_OPEN_HAS_EA)) {
1877                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
1878                                         spec->u.sp_ea.eadatalen);
1879         } else {
1880                 buf = &LU_BUF_NULL;
1881         }
1882
1883         rc = dt_declare_xattr_set(env, mdd_object_child(son), buf,
1884                                   XATTR_NAME_LOV, 0, handle);
1885         if (rc)
1886                 GOTO(stop, rc);
1887
1888         rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL, NULL,
1889                                          handle);
1890         if (rc)
1891                 GOTO(stop, rc);
1892
1893         rc = mdd_trans_start(env, mdd, handle);
1894         if (rc)
1895                 GOTO(stop, rc);
1896
1897         rc = dt_xattr_set(env, mdd_object_child(son), buf, XATTR_NAME_LOV,
1898                           0, handle);
1899
1900         if (rc)
1901                 GOTO(stop, rc);
1902
1903         rc = mdd_changelog_data_store(env, mdd, CL_LAYOUT, 0, son, handle);
1904
1905 stop:
1906         rc = mdd_trans_stop(env, mdd, rc, handle);
1907
1908 out_free:
1909         RETURN(rc);
1910 }
1911
1912 static int mdd_declare_object_initialize(const struct lu_env *env,
1913                                          struct mdd_object *parent,
1914                                          struct mdd_object *child,
1915                                          const struct lu_attr *attr,
1916                                          struct thandle *handle)
1917 {
1918         int rc;
1919         ENTRY;
1920
1921         LASSERT(attr->la_valid & (LA_MODE | LA_TYPE));
1922         if (!S_ISDIR(attr->la_mode))
1923                 RETURN(0);
1924
1925         rc = mdo_declare_index_insert(env, child, mdo2fid(child), S_IFDIR,
1926                                       dot, handle);
1927         if (rc != 0)
1928                 RETURN(rc);
1929
1930         rc = mdo_declare_ref_add(env, child, handle);
1931         if (rc != 0)
1932                 RETURN(rc);
1933
1934         rc = mdo_declare_index_insert(env, child, mdo2fid(parent), S_IFDIR,
1935                                       dotdot, handle);
1936
1937         RETURN(rc);
1938 }
1939
1940 static int mdd_object_initialize(const struct lu_env *env,
1941                                  const struct lu_fid *pfid,
1942                                  struct mdd_object *child,
1943                                  struct lu_attr *attr,
1944                                  struct thandle *handle)
1945 {
1946         int rc = 0;
1947         ENTRY;
1948
1949         if (S_ISDIR(attr->la_mode)) {
1950                 /* Add "." and ".." for newly created dir */
1951                 mdo_ref_add(env, child, handle);
1952                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1953                                              S_IFDIR, dot, handle);
1954                 if (rc == 0)
1955                         rc = __mdd_index_insert_only(env, child, pfid, S_IFDIR,
1956                                                      dotdot, handle);
1957                 if (rc != 0)
1958                         mdo_ref_del(env, child, handle);
1959         }
1960
1961         RETURN(rc);
1962 }
1963
1964 /**
1965  * This function checks whether it can create a file/dir under the
1966  * directory(@pobj). The directory(@pobj) is not being locked by
1967  * mdd lock.
1968  *
1969  * \param[in] env       execution environment
1970  * \param[in] pobj      the directory to create files
1971  * \param[in] pattr     the attributes of the directory
1972  * \param[in] lname     the name of the created file/dir
1973  * \param[in] cattr     the attributes of the file/dir
1974  * \param[in] spec      create specification
1975  *
1976  * \retval              = 0 it is allowed to create file/dir under
1977  *                      the directory
1978  * \retval              negative error not allowed to create file/dir
1979  *                      under the directory
1980  */
1981 static int mdd_create_sanity_check(const struct lu_env *env,
1982                                    struct md_object *pobj,
1983                                    const struct lu_attr *pattr,
1984                                    const struct lu_name *lname,
1985                                    struct lu_attr *cattr,
1986                                    struct md_op_spec *spec)
1987 {
1988         struct mdd_thread_info *info = mdd_env_info(env);
1989         struct lu_fid     *fid       = &info->mti_fid;
1990         struct mdd_object *obj       = md2mdd_obj(pobj);
1991         struct mdd_device *m         = mdo2mdd(pobj);
1992         bool            check_perm = true;
1993         int rc;
1994         ENTRY;
1995
1996         /* EEXIST check */
1997         if (mdd_is_dead_obj(obj))
1998                 RETURN(-ENOENT);
1999
2000         /*
2001          * In some cases this lookup is not needed - we know before if name
2002          * exists or not because MDT performs lookup for it.
2003          * name length check is done in lookup.
2004          */
2005         if (spec->sp_cr_lookup) {
2006                 /*
2007                  * Check if the name already exist, though it will be checked in
2008                  * _index_insert also, for avoiding rolling back if exists
2009                  * _index_insert.
2010                  */
2011                 rc = __mdd_lookup(env, pobj, pattr, lname, fid,
2012                                   MAY_WRITE | MAY_EXEC);
2013                 if (rc != -ENOENT)
2014                         RETURN(rc ? : -EEXIST);
2015
2016                 /* Permission is already being checked in mdd_lookup */
2017                 check_perm = false;
2018         }
2019
2020         if (S_ISDIR(cattr->la_mode) &&
2021             unlikely(spec != NULL && spec->sp_cr_flags & MDS_OPEN_HAS_EA) &&
2022             spec->u.sp_ea.eadata != NULL && spec->u.sp_ea.eadatalen > 0) {
2023                 const struct lmv_user_md *lum = spec->u.sp_ea.eadata;
2024
2025                 if (!lmv_magic_supported(le32_to_cpu(lum->lum_magic)) &&
2026                     le32_to_cpu(lum->lum_magic) != LMV_USER_MAGIC_V0) {
2027                         rc = -EINVAL;
2028                         CERROR("%s: invalid lmv_user_md: magic = %x, "
2029                                "stripe_offset = %d, stripe_count = %u: "
2030                                "rc = %d\n", mdd2obd_dev(m)->obd_name,
2031                                 le32_to_cpu(lum->lum_magic),
2032                                (int)le32_to_cpu(lum->lum_stripe_offset),
2033                                le32_to_cpu(lum->lum_stripe_count), rc);
2034                         return rc;
2035                 }
2036         }
2037
2038         rc = mdd_may_create(env, obj, pattr, NULL, check_perm);
2039         if (rc != 0)
2040                 RETURN(rc);
2041
2042         /* sgid check */
2043         if (pattr->la_mode & S_ISGID) {
2044                 struct lu_ucred *uc = lu_ucred(env);
2045
2046                 cattr->la_gid = pattr->la_gid;
2047
2048                 /* Directories are special, and always inherit S_ISGID */
2049                 if (S_ISDIR(cattr->la_mode)) {
2050                         cattr->la_mode |= S_ISGID;
2051                         cattr->la_valid |= LA_MODE;
2052                 } else if ((cattr->la_mode & (S_ISGID | S_IXGRP))
2053                                 == (S_ISGID | S_IXGRP) &&
2054                            !lustre_in_group_p(uc,
2055                                               (cattr->la_valid & LA_GID) ?
2056                                               cattr->la_gid : pattr->la_gid) &&
2057                            !md_capable(uc, CFS_CAP_FSETID)) {
2058                         cattr->la_mode &= ~S_ISGID;
2059                         cattr->la_valid |= LA_MODE;
2060                 }
2061         }
2062
2063         /* Inherit project ID from parent directory */
2064         if (pattr->la_flags & LUSTRE_PROJINHERIT_FL) {
2065                 cattr->la_projid = pattr->la_projid;
2066                 if (S_ISDIR(cattr->la_mode)) {
2067                         cattr->la_flags |= LUSTRE_PROJINHERIT_FL;
2068                         cattr->la_valid |= LA_FLAGS;
2069                 }
2070                 cattr->la_valid |= LA_PROJID;
2071         }
2072
2073         rc = mdd_name_check(m, lname);
2074         if (rc < 0)
2075                 RETURN(rc);
2076
2077         switch (cattr->la_mode & S_IFMT) {
2078         case S_IFLNK: {
2079                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
2080
2081                 if (symlen > m->mdd_dt_conf.ddp_symlink_max)
2082                         RETURN(-ENAMETOOLONG);
2083                 else
2084                         RETURN(0);
2085         }
2086         case S_IFDIR:
2087         case S_IFREG:
2088         case S_IFCHR:
2089         case S_IFBLK:
2090         case S_IFIFO:
2091         case S_IFSOCK:
2092                 rc = 0;
2093                 break;
2094         default:
2095                 rc = -EINVAL;
2096                 break;
2097         }
2098         RETURN(rc);
2099 }
2100
2101 static int mdd_declare_create_object(const struct lu_env *env,
2102                                      struct mdd_device *mdd,
2103                                      struct mdd_object *p, struct mdd_object *c,
2104                                      struct lu_attr *attr,
2105                                      struct thandle *handle,
2106                                      const struct md_op_spec *spec,
2107                                      struct lu_buf *def_acl_buf,
2108                                      struct lu_buf *acl_buf,
2109                                      struct lu_buf *hsm_buf,
2110                                      struct dt_allocation_hint *hint)
2111 {
2112         const struct lu_buf *buf;
2113         int rc;
2114
2115         rc = mdd_declare_create_object_internal(env, p, c, attr, handle, spec,
2116                                                 hint);
2117         if (rc)
2118                 GOTO(out, rc);
2119
2120 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
2121         if (def_acl_buf && def_acl_buf->lb_len > 0 && S_ISDIR(attr->la_mode)) {
2122                 /* if dir, then can inherit default ACl */
2123                 rc = mdo_declare_xattr_set(env, c, def_acl_buf,
2124                                            XATTR_NAME_ACL_DEFAULT,
2125                                            0, handle);
2126                 if (rc)
2127                         GOTO(out, rc);
2128         }
2129
2130         if (acl_buf && acl_buf->lb_len > 0) {
2131                 rc = mdo_declare_attr_set(env, c, attr, handle);
2132                 if (rc)
2133                         GOTO(out, rc);
2134
2135                 rc = mdo_declare_xattr_set(env, c, acl_buf,
2136                                            XATTR_NAME_ACL_ACCESS, 0, handle);
2137                 if (rc)
2138                         GOTO(out, rc);
2139         }
2140 #endif
2141         rc = mdd_declare_object_initialize(env, p, c, attr, handle);
2142         if (rc)
2143                 GOTO(out, rc);
2144
2145         /* replay case, create LOV EA from client data */
2146         if ((!(spec->sp_cr_flags & MDS_OPEN_DELAY_CREATE) && spec->no_create) ||
2147             (spec->sp_cr_flags & MDS_OPEN_HAS_EA && S_ISREG(attr->la_mode))) {
2148                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2149                                         spec->u.sp_ea.eadatalen);
2150                 rc = mdo_declare_xattr_set(env, c, buf,
2151                                            S_ISDIR(attr->la_mode) ?
2152                                                 XATTR_NAME_LMV : XATTR_NAME_LOV,
2153                                            0, handle);
2154                 if (rc)
2155                         GOTO(out, rc);
2156
2157                 if (spec->sp_cr_flags & MDS_OPEN_PCC) {
2158                         rc = mdo_declare_xattr_set(env, c, hsm_buf,
2159                                                    XATTR_NAME_HSM,
2160                                                    0, handle);
2161                         if (rc)
2162                                 GOTO(out, rc);
2163                 }
2164         }
2165
2166         if (S_ISLNK(attr->la_mode)) {
2167                 const char *target_name = spec->u.sp_symname;
2168                 int sym_len = strlen(target_name);
2169                 const struct lu_buf *buf;
2170
2171                 buf = mdd_buf_get_const(env, target_name, sym_len);
2172                 rc = dt_declare_record_write(env, mdd_object_child(c),
2173                                              buf, 0, handle);
2174                 if (rc)
2175                         GOTO(out, rc);
2176         }
2177
2178         if (spec->sp_cr_file_secctx_name != NULL) {
2179                 buf = mdd_buf_get_const(env, spec->sp_cr_file_secctx,
2180                                         spec->sp_cr_file_secctx_size);
2181                 rc = mdo_declare_xattr_set(env, c, buf,
2182                                            spec->sp_cr_file_secctx_name, 0,
2183                                            handle);
2184                 if (rc < 0)
2185                         GOTO(out, rc);
2186         }
2187 out:
2188         return rc;
2189 }
2190
2191 static int mdd_declare_create(const struct lu_env *env, struct mdd_device *mdd,
2192                               struct mdd_object *p, struct mdd_object *c,
2193                               const struct lu_name *name,
2194                               struct lu_attr *attr,
2195                               struct thandle *handle,
2196                               const struct md_op_spec *spec,
2197                               struct linkea_data *ldata,
2198                               struct lu_buf *def_acl_buf,
2199                               struct lu_buf *acl_buf,
2200                               struct lu_buf *hsm_buf,
2201                               struct dt_allocation_hint *hint)
2202 {
2203         int rc;
2204
2205         rc = mdd_declare_create_object(env, mdd, p, c, attr, handle, spec,
2206                                        def_acl_buf, acl_buf, hsm_buf, hint);
2207         if (rc)
2208                 GOTO(out, rc);
2209
2210         if (S_ISDIR(attr->la_mode)) {
2211                 rc = mdo_declare_ref_add(env, p, handle);
2212                 if (rc)
2213                         GOTO(out, rc);
2214         }
2215
2216         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2217                 rc = mdd_orphan_declare_insert(env, c, attr->la_mode, handle);
2218                 if (rc)
2219                         GOTO(out, rc);
2220         } else {
2221                 struct lu_attr *la = &mdd_env_info(env)->mti_la_for_fix;
2222                 enum changelog_rec_type type;
2223
2224                 rc = mdo_declare_index_insert(env, p, mdo2fid(c), attr->la_mode,
2225                                               name->ln_name, handle);
2226                 if (rc != 0)
2227                         return rc;
2228
2229                 rc = mdd_declare_links_add(env, c, handle, ldata);
2230                 if (rc)
2231                         return rc;
2232
2233                 *la = *attr;
2234                 la->la_valid = LA_CTIME | LA_MTIME;
2235                 rc = mdo_declare_attr_set(env, p, la, handle);
2236                 if (rc)
2237                         return rc;
2238
2239                 type = S_ISDIR(attr->la_mode) ? CL_MKDIR :
2240                        S_ISREG(attr->la_mode) ? CL_CREATE :
2241                        S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD;
2242
2243                 rc = mdd_declare_changelog_store(env, mdd, type, name, NULL,
2244                                                  handle);
2245                 if (rc)
2246                         return rc;
2247         }
2248 out:
2249         return rc;
2250 }
2251
2252 static int mdd_acl_init(const struct lu_env *env, struct mdd_object *pobj,
2253                         struct lu_attr *la, struct lu_buf *def_acl_buf,
2254                         struct lu_buf *acl_buf)
2255 {
2256         int     rc;
2257         ENTRY;
2258
2259         if (S_ISLNK(la->la_mode)) {
2260                 acl_buf->lb_len = 0;
2261                 def_acl_buf->lb_len = 0;
2262                 RETURN(0);
2263         }
2264
2265         mdd_read_lock(env, pobj, DT_TGT_PARENT);
2266         rc = mdo_xattr_get(env, pobj, def_acl_buf,
2267                            XATTR_NAME_ACL_DEFAULT);
2268         mdd_read_unlock(env, pobj);
2269         if (rc > 0) {
2270                 /* If there are default ACL, fix mode/ACL by default ACL */
2271                 def_acl_buf->lb_len = rc;
2272                 LASSERT(def_acl_buf->lb_len <= acl_buf->lb_len);
2273                 memcpy(acl_buf->lb_buf, def_acl_buf->lb_buf, rc);
2274                 acl_buf->lb_len = rc;
2275                 rc = __mdd_fix_mode_acl(env, acl_buf, &la->la_mode);
2276                 if (rc < 0)
2277                         RETURN(rc);
2278         } else if (rc == -ENODATA || rc == -EOPNOTSUPP) {
2279                 /* If there are no default ACL, fix mode by mask */
2280                 struct lu_ucred *uc = lu_ucred(env);
2281
2282                 /* The create triggered by MDT internal events, such as
2283                  * LFSCK reset, will not contain valid "uc". */
2284                 if (unlikely(uc != NULL))
2285                         la->la_mode &= ~uc->uc_umask;
2286                 rc = 0;
2287                 acl_buf->lb_len = 0;
2288                 def_acl_buf->lb_len = 0;
2289         }
2290
2291         RETURN(rc);
2292 }
2293
2294 /**
2295  * Create a metadata object and initialize it, set acl, xattr.
2296  **/
2297 static int mdd_create_object(const struct lu_env *env, struct mdd_object *pobj,
2298                              struct mdd_object *son, struct lu_attr *attr,
2299                              struct md_op_spec *spec, struct lu_buf *acl_buf,
2300                              struct lu_buf *def_acl_buf,
2301                              struct lu_buf *hsm_buf,
2302                              struct dt_allocation_hint *hint,
2303                              struct thandle *handle)
2304 {
2305         const struct lu_buf *buf;
2306         int rc;
2307
2308         mdd_write_lock(env, son, DT_TGT_CHILD);
2309         rc = mdd_create_object_internal(env, NULL, son, attr, handle, spec,
2310                                         hint);
2311         if (rc)
2312                 GOTO(unlock, rc);
2313
2314         /* Note: In DNE phase I, for striped dir, though sub-stripes will be
2315          * created in declare phase, they also needs to be added to master
2316          * object as sub-directory entry. So it has to initialize the master
2317          * object, then set dir striped EA.(in mdo_xattr_set) */
2318         rc = mdd_object_initialize(env, mdo2fid(pobj), son, attr, handle);
2319         if (rc != 0)
2320                 GOTO(err_destroy, rc);
2321
2322         /*
2323          * in case of replay we just set LOVEA provided by the client
2324          * XXX: I think it would be interesting to try "old" way where
2325          *      MDT calls this xattr_set(LOV) in a different transaction.
2326          *      probably this way we code can be made better.
2327          */
2328
2329         /* During creation, there are only a few cases we need do xattr_set to
2330          * create stripes.
2331          * 1. regular file: see comments above.
2332          * 2. dir: inherit default striping or pool settings from parent.
2333          * 3. create striped directory with provided stripeEA.
2334          * 4. create striped directory because inherit default layout from the
2335          * parent.
2336          */
2337         if (spec->no_create ||
2338             (S_ISREG(attr->la_mode) && spec->sp_cr_flags & MDS_OPEN_HAS_EA) ||
2339             S_ISDIR(attr->la_mode)) {
2340                 buf = mdd_buf_get_const(env, spec->u.sp_ea.eadata,
2341                                         spec->u.sp_ea.eadatalen);
2342                 rc = mdo_xattr_set(env, son, buf,
2343                                    S_ISDIR(attr->la_mode) ? XATTR_NAME_LMV :
2344                                                             XATTR_NAME_LOV,
2345                                    0, handle);
2346                 if (rc != 0)
2347                         GOTO(err_destroy, rc);
2348         }
2349
2350         if (S_ISREG(attr->la_mode) && spec->sp_cr_flags & MDS_OPEN_PCC) {
2351                 struct md_hsm mh;
2352
2353                 memset(&mh, 0, sizeof(mh));
2354                 mh.mh_flags = HS_EXISTS | HS_ARCHIVED | HS_RELEASED;
2355                 mh.mh_arch_id = spec->sp_archive_id;
2356                 lustre_hsm2buf(hsm_buf->lb_buf, &mh);
2357                 rc = mdo_xattr_set(env, son, hsm_buf, XATTR_NAME_HSM,
2358                                    0, handle);
2359                 if (rc != 0)
2360                         GOTO(err_destroy, rc);
2361         }
2362
2363 #ifdef CONFIG_LUSTRE_FS_POSIX_ACL
2364         if (def_acl_buf != NULL && def_acl_buf->lb_len > 0 &&
2365             S_ISDIR(attr->la_mode)) {
2366                 /* set default acl */
2367                 rc = mdo_xattr_set(env, son, def_acl_buf,
2368                                    XATTR_NAME_ACL_DEFAULT, 0,
2369                                    handle);
2370                 if (rc)
2371                         GOTO(err_destroy, rc);
2372         }
2373         /* set its own acl */
2374         if (acl_buf != NULL && acl_buf->lb_len > 0) {
2375                 rc = mdo_xattr_set(env, son, acl_buf,
2376                                    XATTR_NAME_ACL_ACCESS,
2377                                    0, handle);
2378                 if (rc)
2379                         GOTO(err_destroy, rc);
2380         }
2381 #endif
2382
2383         if (S_ISLNK(attr->la_mode)) {
2384                 struct dt_object *dt = mdd_object_child(son);
2385                 const char *target_name = spec->u.sp_symname;
2386                 int sym_len = strlen(target_name);
2387                 loff_t pos = 0;
2388
2389                 buf = mdd_buf_get_const(env, target_name, sym_len);
2390                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle);
2391                 if (rc == sym_len)
2392                         rc = 0;
2393                 else
2394                         GOTO(err_initlized, rc = -EFAULT);
2395         }
2396
2397         if (spec->sp_cr_file_secctx_name != NULL) {
2398                 buf = mdd_buf_get_const(env, spec->sp_cr_file_secctx,
2399                                         spec->sp_cr_file_secctx_size);
2400                 rc = mdo_xattr_set(env, son, buf, spec->sp_cr_file_secctx_name,
2401                                    0, handle);
2402                 if (rc < 0)
2403                         GOTO(err_initlized, rc);
2404         }
2405
2406 err_initlized:
2407         if (unlikely(rc != 0)) {
2408                 int rc2;
2409                 if (S_ISDIR(attr->la_mode)) {
2410                         /* Drop the reference, no need to delete "."/"..",
2411                          * because the object to be destroied directly. */
2412                         rc2 = mdo_ref_del(env, son, handle);
2413                         if (rc2 != 0)
2414                                 GOTO(unlock, rc);
2415                 }
2416                 rc2 = mdo_ref_del(env, son, handle);
2417                 if (rc2 != 0)
2418                         GOTO(unlock, rc);
2419 err_destroy:
2420                 mdo_destroy(env, son, handle);
2421         }
2422 unlock:
2423         mdd_write_unlock(env, son);
2424         RETURN(rc);
2425 }
2426
2427 static int mdd_index_delete(const struct lu_env *env,
2428                             struct mdd_object *mdd_pobj,
2429                             struct lu_attr *cattr,
2430                             const struct lu_name *lname)
2431 {
2432         struct mdd_device *mdd = mdo2mdd(&mdd_pobj->mod_obj);
2433         struct thandle *handle;
2434         int rc;
2435         ENTRY;
2436
2437         handle = mdd_trans_create(env, mdd);
2438         if (IS_ERR(handle))
2439                 RETURN(PTR_ERR(handle));
2440
2441         rc = mdo_declare_index_delete(env, mdd_pobj, lname->ln_name,
2442                                       handle);
2443         if (rc != 0)
2444                 GOTO(stop, rc);
2445
2446         if (S_ISDIR(cattr->la_mode)) {
2447                 rc = mdo_declare_ref_del(env, mdd_pobj, handle);
2448                 if (rc != 0)
2449                         GOTO(stop, rc);
2450         }
2451
2452         /* Since this will only be used in the error handler path,
2453          * Let's set the thandle to be local and not mess the transno */
2454         handle->th_local = 1;
2455         rc = mdd_trans_start(env, mdd, handle);
2456         if (rc)
2457                 GOTO(stop, rc);
2458
2459         rc = __mdd_index_delete(env, mdd_pobj, lname->ln_name,
2460                                 S_ISDIR(cattr->la_mode), handle);
2461         if (rc)
2462                 GOTO(stop, rc);
2463 stop:
2464         rc = mdd_trans_stop(env, mdd, rc, handle);
2465
2466         RETURN(rc);
2467 }
2468
2469 /**
2470  * Create object and insert it into namespace.
2471  *
2472  * Two operations have to be performed:
2473  *
2474  *  - an allocation of a new object (->do_create()), and
2475  *  - an insertion into a parent index (->dio_insert()).
2476  *
2477  * Due to locking, operation order is not important, when both are
2478  * successful, *but* error handling cases are quite different:
2479  *
2480  *  - if insertion is done first, and following object creation fails,
2481  *  insertion has to be rolled back, but this operation might fail
2482  *  also leaving us with dangling index entry.
2483  *
2484  *  - if creation is done first, is has to be undone if insertion fails,
2485  *  leaving us with leaked space, which is not good but not fatal.
2486  *
2487  * It seems that creation-first is simplest solution, but it is sub-optimal
2488  * in the frequent
2489  *
2490  * $ mkdir foo
2491  * $ mkdir foo
2492  *
2493  * case, because second mkdir is bound to create object, only to
2494  * destroy it immediately.
2495  *
2496  * To avoid this follow local file systems that do double lookup:
2497  *
2498  * 0. lookup -> -EEXIST (mdd_create_sanity_check())
2499  * 1. create            (mdd_create_object_internal())
2500  * 2. insert            (__mdd_index_insert(), lookup again)
2501  *
2502  * \param[in] pobj      parent object
2503  * \param[in] lname     name of child being created
2504  * \param[in,out] child child object being created
2505  * \param[in] spec      additional create parameters
2506  * \param[in] ma        attributes for new child object
2507  *
2508  * \retval              0 on success
2509  * \retval              negative errno on failure
2510  */
2511 int mdd_create(const struct lu_env *env, struct md_object *pobj,
2512                       const struct lu_name *lname, struct md_object *child,
2513                       struct md_op_spec *spec, struct md_attr *ma)
2514 {
2515         struct mdd_thread_info  *info = mdd_env_info(env);
2516         struct lu_attr          *la = &info->mti_la_for_fix;
2517         struct mdd_object       *mdd_pobj = md2mdd_obj(pobj);
2518         struct mdd_object       *son = md2mdd_obj(child);
2519         struct mdd_device       *mdd = mdo2mdd(pobj);
2520         struct lu_attr          *attr = &ma->ma_attr;
2521         struct thandle          *handle;
2522         struct lu_attr          *pattr = &info->mti_pattr;
2523         struct lu_buf           acl_buf;
2524         struct lu_buf           def_acl_buf;
2525         struct lu_buf           hsm_buf;
2526         struct linkea_data      *ldata = &info->mti_link_data;
2527         const char              *name = lname->ln_name;
2528         struct dt_allocation_hint *hint = &mdd_env_info(env)->mti_hint;
2529         int                      rc;
2530         int                      rc2;
2531         ENTRY;
2532
2533         rc = mdd_la_get(env, mdd_pobj, pattr);
2534         if (rc != 0)
2535                 RETURN(rc);
2536
2537         /* Sanity checks before big job. */
2538         rc = mdd_create_sanity_check(env, pobj, pattr, lname, attr, spec);
2539         if (rc)
2540                 RETURN(rc);
2541
2542         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_DQACQ_NET))
2543                 GOTO(out_free, rc = -EINPROGRESS);
2544
2545         handle = mdd_trans_create(env, mdd);
2546         if (IS_ERR(handle))
2547                 GOTO(out_free, rc = PTR_ERR(handle));
2548
2549         lu_buf_check_and_alloc(&info->mti_xattr_buf,
2550                         MIN(mdd->mdd_dt_conf.ddp_max_ea_size, XATTR_SIZE_MAX));
2551         acl_buf = info->mti_xattr_buf;
2552         def_acl_buf.lb_buf = info->mti_key;
2553         def_acl_buf.lb_len = sizeof(info->mti_key);
2554         rc = mdd_acl_init(env, mdd_pobj, attr, &def_acl_buf, &acl_buf);
2555         if (rc < 0)
2556                 GOTO(out_stop, rc);
2557
2558         if (S_ISDIR(attr->la_mode)) {
2559                 struct lmv_user_md *lmu = spec->u.sp_ea.eadata;
2560
2561                 /*
2562                  * migrate may create 1-stripe directory, so lod_ah_init()
2563                  * doesn't adjust stripe count from lmu.
2564                  */
2565                 if (lmu && lmu->lum_stripe_count == cpu_to_le32(1)) {
2566                         info->mti_lmu = *lmu;
2567                         info->mti_lmu.lum_stripe_count = 0;
2568                         spec->u.sp_ea.eadata = &info->mti_lmu;
2569                 }
2570         }
2571
2572         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
2573
2574         memset(ldata, 0, sizeof(*ldata));
2575         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
2576                 struct lu_fid tfid = *mdd_object_fid(mdd_pobj);
2577
2578                 tfid.f_oid--;
2579                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2580                                         &tfid, lname, 1, 0, ldata);
2581         } else {
2582                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2583                                         mdd_object_fid(mdd_pobj),
2584                                         lname, 1, 0, ldata);
2585         }
2586
2587         if (spec->sp_cr_flags & MDS_OPEN_PCC) {
2588                 LASSERT(spec->sp_cr_flags & MDS_OPEN_HAS_EA);
2589
2590                 memset(&hsm_buf, 0, sizeof(hsm_buf));
2591                 lu_buf_alloc(&hsm_buf, sizeof(struct hsm_attrs));
2592                 if (hsm_buf.lb_buf == NULL)
2593                         GOTO(out_stop, rc = -ENOMEM);
2594         }
2595
2596         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, attr,
2597                                 handle, spec, ldata, &def_acl_buf, &acl_buf,
2598                                 &hsm_buf, hint);
2599         if (rc)
2600                 GOTO(out_stop, rc);
2601
2602         rc = mdd_trans_start(env, mdd, handle);
2603         if (rc)
2604                 GOTO(out_stop, rc);
2605
2606         rc = mdd_create_object(env, mdd_pobj, son, attr, spec, &acl_buf,
2607                                &def_acl_buf, &hsm_buf, hint, handle);
2608         if (rc != 0)
2609                 GOTO(out_stop, rc);
2610
2611         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2612                 mdd_write_lock(env, son, DT_TGT_CHILD);
2613                 son->mod_flags |= VOLATILE_OBJ;
2614                 rc = mdd_orphan_insert(env, son, handle);
2615                 GOTO(out_volatile, rc);
2616         } else {
2617                 rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2618                                       attr->la_mode, name, handle);
2619                 if (rc != 0)
2620                         GOTO(err_created, rc);
2621
2622                 mdd_links_add(env, son, mdo2fid(mdd_pobj), lname, handle,
2623                               ldata, 1);
2624
2625                 /* update parent directory mtime/ctime */
2626                 *la = *attr;
2627                 la->la_valid = LA_CTIME | LA_MTIME;
2628                 rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
2629                 if (rc)
2630                         GOTO(err_insert, rc);
2631         }
2632
2633         EXIT;
2634 err_insert:
2635         if (rc != 0) {
2636                 if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
2637                         rc2 = mdd_orphan_delete(env, son, handle);
2638                 else
2639                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2640                                                  S_ISDIR(attr->la_mode),
2641                                                  handle);
2642                 if (rc2 != 0)
2643                         goto out_stop;
2644
2645 err_created:
2646                 mdd_write_lock(env, son, DT_TGT_CHILD);
2647                 if (S_ISDIR(attr->la_mode)) {
2648                         /* Drop the reference, no need to delete "."/"..",
2649                          * because the object is to be destroyed directly. */
2650                         rc2 = mdo_ref_del(env, son, handle);
2651                         if (rc2 != 0) {
2652                                 mdd_write_unlock(env, son);
2653                                 goto out_stop;
2654                         }
2655                 }
2656 out_volatile:
2657                 /* For volatile files drop one link immediately, since there is
2658                  * no filename in the namespace, and save any error returned. */
2659                 rc2 = mdo_ref_del(env, son, handle);
2660                 if (rc2 != 0) {
2661                         mdd_write_unlock(env, son);
2662                         if (unlikely(rc == 0))
2663                                 rc = rc2;
2664                         goto out_stop;
2665                 }
2666
2667                 /* Don't destroy the volatile object on success */
2668                 if (likely(rc != 0))
2669                         mdo_destroy(env, son, handle);
2670                 mdd_write_unlock(env, son);
2671         }
2672
2673         if (rc == 0 && fid_is_namespace_visible(mdo2fid(son)) &&
2674             likely((spec->sp_cr_flags & MDS_OPEN_VOLATILE) == 0))
2675                 rc = mdd_changelog_ns_store(env, mdd,
2676                                 S_ISDIR(attr->la_mode) ? CL_MKDIR :
2677                                 S_ISREG(attr->la_mode) ? CL_CREATE :
2678                                 S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2679                                 0, son, mdo2fid(mdd_pobj), NULL, NULL, lname,
2680                                 NULL, handle);
2681 out_stop:
2682         rc2 = mdd_trans_stop(env, mdd, rc, handle);
2683         if (rc == 0) {
2684                 /* If creation fails, it is most likely due to the remote update
2685                  * failure, because local transaction will mostly succeed at
2686                  * this stage. There is no easy way to rollback all of previous
2687                  * updates, so let's remove the object from namespace, and
2688                  * LFSCK should handle the orphan object. */
2689                 if (rc2 < 0 && !mdd_object_remote(mdd_pobj))
2690                         mdd_index_delete(env, mdd_pobj, attr, lname);
2691                 rc = rc2;
2692         }
2693 out_free:
2694         if (is_vmalloc_addr(ldata->ld_buf))
2695                 /* if we vmalloced a large buffer drop it */
2696                 lu_buf_free(ldata->ld_buf);
2697
2698         if (spec->sp_cr_flags & MDS_OPEN_PCC)
2699                 lu_buf_free(&hsm_buf);
2700
2701         /* The child object shouldn't be cached anymore */
2702         if (rc)
2703                 set_bit(LU_OBJECT_HEARD_BANSHEE,
2704                         &child->mo_lu.lo_header->loh_flags);
2705         return rc;
2706 }
2707
2708 /*
2709  * Get locks on parents in proper order
2710  * RETURN: < 0 - error, rename_order if successful
2711  */
2712 enum rename_order {
2713         MDD_RN_SAME,
2714         MDD_RN_SRCTGT,
2715         MDD_RN_TGTSRC
2716 };
2717
2718 static int mdd_rename_order(const struct lu_env *env,
2719                             struct mdd_device *mdd,
2720                             struct mdd_object *src_pobj,
2721                             const struct lu_attr *pattr,
2722                             struct mdd_object *tgt_pobj)
2723 {
2724         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2725         int rc;
2726         ENTRY;
2727
2728         if (src_pobj == tgt_pobj)
2729                 RETURN(MDD_RN_SAME);
2730
2731         /* compared the parent child relationship of src_p&tgt_p */
2732         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2733                 rc = MDD_RN_SRCTGT;
2734         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2735                 rc = MDD_RN_TGTSRC;
2736         } else {
2737                 rc = mdd_is_parent(env, mdd, src_pobj, pattr,
2738                                    mdo2fid(tgt_pobj));
2739                 if (rc == -EREMOTE)
2740                         rc = 0;
2741
2742                 if (rc == 1)
2743                         rc = MDD_RN_TGTSRC;
2744                 else if (rc == 0)
2745                         rc = MDD_RN_SRCTGT;
2746         }
2747
2748         RETURN(rc);
2749 }
2750
2751 /* has not mdd_write{read}_lock on any obj yet. */
2752 static int mdd_rename_sanity_check(const struct lu_env *env,
2753                                    struct mdd_object *src_pobj,
2754                                    const struct lu_attr *pattr,
2755                                    struct mdd_object *tgt_pobj,
2756                                    const struct lu_attr *tpattr,
2757                                    struct mdd_object *sobj,
2758                                    const struct lu_attr *cattr,
2759                                    struct mdd_object *tobj,
2760                                    const struct lu_attr *tattr)
2761 {
2762         int rc = 0;
2763         ENTRY;
2764
2765         /* XXX: when get here, sobj must NOT be NULL,
2766          * the other case has been processed in cld_rename
2767          * before mdd_rename and enable MDS_PERM_BYPASS. */
2768         LASSERT(sobj);
2769
2770         /*
2771          * If we are using project inheritance, we only allow renames
2772          * into our tree when the project IDs are the same; otherwise
2773          * tree quota mechanism would be circumvented.
2774          */
2775         if (((tpattr->la_flags & LUSTRE_PROJINHERIT_FL) &&
2776             tpattr->la_projid != cattr->la_projid) ||
2777             ((pattr->la_flags & LUSTRE_PROJINHERIT_FL) &&
2778             (pattr->la_projid != tpattr->la_projid)))
2779                 RETURN(-EXDEV);
2780
2781         rc = mdd_may_delete(env, src_pobj, pattr, sobj, cattr, NULL, 1, 0);
2782         if (rc)
2783                 RETURN(rc);
2784
2785         /* XXX: when get here, "tobj == NULL" means tobj must
2786          * NOT exist (neither on remote MDS, such case has been
2787          * processed in cld_rename before mdd_rename and enable
2788          * MDS_PERM_BYPASS).
2789          * So check may_create, but not check may_unlink. */
2790         if (tobj == NULL)
2791                 rc = mdd_may_create(env, tgt_pobj, tpattr, NULL,
2792                                     (src_pobj != tgt_pobj));
2793         else
2794                 rc = mdd_may_delete(env, tgt_pobj, tpattr, tobj, tattr, cattr,
2795                                     (src_pobj != tgt_pobj), 1);
2796
2797         if (!rc && !tobj && (src_pobj != tgt_pobj) && S_ISDIR(cattr->la_mode))
2798                 rc = __mdd_may_link(env, tgt_pobj, tpattr);
2799
2800         RETURN(rc);
2801 }
2802
2803 static int mdd_declare_rename(const struct lu_env *env,
2804                               struct mdd_device *mdd,
2805                               struct mdd_object *mdd_spobj,
2806                               struct mdd_object *mdd_tpobj,
2807                               struct mdd_object *mdd_sobj,
2808                               struct mdd_object *mdd_tobj,
2809                               const struct lu_name *sname,
2810                               const struct lu_name *tname,
2811                               struct md_attr *ma,
2812                               struct linkea_data *ldata,
2813                               struct thandle *handle)
2814 {
2815         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2816         int rc;
2817
2818         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2819         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2820
2821         LASSERT(mdd_spobj);
2822         LASSERT(mdd_tpobj);
2823         LASSERT(mdd_sobj);
2824
2825         /* name from source dir */
2826         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2827         if (rc)
2828                 return rc;
2829
2830         /* .. from source child */
2831         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2832                 /* source child can be directory,
2833                  * counted by source dir's nlink */
2834                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2835                 if (rc)
2836                         return rc;
2837                 if (mdd_spobj != mdd_tpobj) {
2838                         rc = mdo_declare_index_delete(env, mdd_sobj, dotdot,
2839                                                       handle);
2840                         if (rc != 0)
2841                                 return rc;
2842
2843                         rc = mdo_declare_index_insert(env, mdd_sobj,
2844                                                       mdo2fid(mdd_tpobj),
2845                                                       S_IFDIR, dotdot, handle);
2846                         if (rc != 0)
2847                                 return rc;
2848                 }
2849
2850                 /* new target child can be directory,
2851                  * counted by target dir's nlink */
2852                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2853                 if (rc != 0)
2854                         return rc;
2855         }
2856
2857         la->la_valid = LA_CTIME | LA_MTIME;
2858         rc = mdo_declare_attr_set(env, mdd_spobj, la, handle);
2859         if (rc != 0)
2860                 return rc;
2861
2862         rc = mdo_declare_attr_set(env, mdd_tpobj, la, handle);
2863         if (rc != 0)
2864                 return rc;
2865
2866         la->la_valid = LA_CTIME;
2867         rc = mdo_declare_attr_set(env, mdd_sobj, la, handle);
2868         if (rc)
2869                 return rc;
2870
2871         rc = mdd_declare_links_add(env, mdd_sobj, handle, ldata);
2872         if (rc)
2873                 return rc;
2874
2875         /* new name */
2876         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2877                                       mdd_object_type(mdd_sobj),
2878                                       tname->ln_name, handle);
2879         if (rc != 0)
2880                 return rc;
2881
2882         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2883                 /* delete target child in target parent directory */
2884                 rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name,
2885                                               handle);
2886                 if (rc)
2887                         return rc;
2888
2889                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2890                 if (rc)
2891                         return rc;
2892
2893                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2894                         /* target child can be directory,
2895                          * delete "." reference in target child directory */
2896                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2897                         if (rc)
2898                                 return rc;
2899
2900                         /* delete ".." reference in target parent directory */
2901                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2902                         if (rc)
2903                                 return rc;
2904                 }
2905
2906                 la->la_valid = LA_CTIME;
2907                 rc = mdo_declare_attr_set(env, mdd_tobj, la, handle);
2908                 if (rc)
2909                         return rc;
2910
2911                 rc = mdd_declare_finish_unlink(env, mdd_tobj, handle);
2912                 if (rc)
2913                         return rc;
2914         }
2915
2916         rc = mdd_declare_changelog_store(env, mdd, CL_RENAME, tname, sname,
2917                                          handle);
2918         if (rc)
2919                 return rc;
2920
2921         return rc;
2922 }
2923
2924 /* src object can be remote that is why we use only fid and type of object */
2925 static int mdd_rename(const struct lu_env *env,
2926                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2927                       const struct lu_fid *lf, const struct lu_name *lsname,
2928                       struct md_object *tobj, const struct lu_name *ltname,
2929                       struct md_attr *ma)
2930 {
2931         const char *sname = lsname->ln_name;
2932         const char *tname = ltname->ln_name;
2933         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2934         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2935         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2936         struct mdd_device *mdd = mdo2mdd(src_pobj);
2937         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2938         struct mdd_object *mdd_tobj = NULL;
2939         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
2940         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
2941         struct lu_attr *tattr = MDD_ENV_VAR(env, tattr);
2942         struct lu_attr *tpattr = MDD_ENV_VAR(env, tpattr);
2943         struct thandle *handle;
2944         struct linkea_data  *ldata = &mdd_env_info(env)->mti_link_data;
2945         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2946         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2947         bool is_dir;
2948         bool tobj_ref = 0;
2949         bool tobj_locked = 0;
2950         unsigned cl_flags = 0;
2951         int rc, rc2;
2952         ENTRY;
2953
2954         /* let unlink to complete and commit */
2955         CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLY_DATA_RACE, 2 + cfs_fail_val);
2956
2957         if (tobj)
2958                 mdd_tobj = md2mdd_obj(tobj);
2959
2960         mdd_sobj = mdd_object_find(env, mdd, lf);
2961         if (IS_ERR(mdd_sobj))
2962                 RETURN(PTR_ERR(mdd_sobj));
2963
2964         rc = mdd_la_get(env, mdd_sobj, cattr);
2965         if (rc)
2966                 GOTO(out_pending, rc);
2967
2968         rc = mdd_la_get(env, mdd_spobj, pattr);
2969         if (rc)
2970                 GOTO(out_pending, rc);
2971
2972         if (mdd_tobj) {
2973                 rc = mdd_la_get(env, mdd_tobj, tattr);
2974                 if (rc)
2975                         GOTO(out_pending, rc);
2976                 /* search for an existing archive.
2977                  * we should check ahead as the object
2978                  * can be destroyed in this transaction */
2979                 if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
2980                         cl_flags |= CLF_RENAME_LAST_EXISTS;
2981         }
2982
2983         rc = mdd_la_get(env, mdd_tpobj, tpattr);
2984         if (rc)
2985                 GOTO(out_pending, rc);
2986
2987         rc = mdd_rename_sanity_check(env, mdd_spobj, pattr, mdd_tpobj, tpattr,
2988                                      mdd_sobj, cattr, mdd_tobj, tattr);
2989         if (rc)
2990                 GOTO(out_pending, rc);
2991
2992         rc = mdd_name_check(mdd, ltname);
2993         if (rc < 0)
2994                 GOTO(out_pending, rc);
2995
2996         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2997         rc = mdd_rename_order(env, mdd, mdd_spobj, pattr, mdd_tpobj);
2998         if (rc < 0)
2999                 GOTO(out_pending, rc);
3000
3001         handle = mdd_trans_create(env, mdd);
3002         if (IS_ERR(handle))
3003                 GOTO(out_pending, rc = PTR_ERR(handle));
3004
3005         memset(ldata, 0, sizeof(*ldata));
3006         rc = mdd_linkea_prepare(env, mdd_sobj, mdd_object_fid(mdd_spobj),
3007                                 lsname, mdd_object_fid(mdd_tpobj), ltname,
3008                                 1, 0, ldata);
3009         if (rc)
3010                 GOTO(stop, rc);
3011
3012         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
3013                                 mdd_tobj, lsname, ltname, ma, ldata, handle);
3014         if (rc)
3015                 GOTO(stop, rc);
3016
3017         rc = mdd_trans_start(env, mdd, handle);
3018         if (rc)
3019                 GOTO(stop, rc);
3020
3021         is_dir = S_ISDIR(cattr->la_mode);
3022
3023         /* Remove source name from source directory */
3024         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle);
3025         if (rc != 0)
3026                 GOTO(stop, rc);
3027
3028         /* "mv dir1 dir2" needs "dir1/.." link update */
3029         if (is_dir && !lu_fid_eq(spobj_fid, tpobj_fid)) {
3030                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3031                 if (rc != 0)
3032                         GOTO(fixup_spobj2, rc);
3033
3034                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, S_IFDIR,
3035                                              dotdot, handle);
3036                 if (rc != 0)
3037                         GOTO(fixup_spobj, rc);
3038         }
3039
3040         if (mdd_tobj != NULL && mdd_object_exists(mdd_tobj)) {
3041                 rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
3042                 if (rc != 0)
3043                         /* tname might been renamed to something else */
3044                         GOTO(fixup_spobj, rc);
3045         }
3046
3047         /* Insert new fid with target name into target dir */
3048         rc = __mdd_index_insert(env, mdd_tpobj, lf, cattr->la_mode,
3049                                 tname, handle);
3050         if (rc != 0)
3051                 GOTO(fixup_tpobj, rc);
3052
3053         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
3054         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
3055
3056         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
3057         la->la_valid = LA_CTIME;
3058         rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
3059         if (rc)
3060                 GOTO(fixup_tpobj, rc);
3061
3062         /* Update the linkEA for the source object */
3063         mdd_write_lock(env, mdd_sobj, DT_SRC_CHILD);
3064         rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
3065                               mdo2fid(mdd_tpobj), ltname, handle, ldata,
3066                               0, 0);
3067         if (rc == -ENOENT)
3068                 /* Old files might not have EA entry */
3069                 mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
3070                               lsname, handle, NULL, 0);
3071         mdd_write_unlock(env, mdd_sobj);
3072         /* We don't fail the transaction if the link ea can't be
3073            updated -- fid2path will use alternate lookup method. */
3074         rc = 0;
3075
3076         /* Remove old target object
3077          * For tobj is remote case cmm layer has processed
3078          * and set tobj to NULL then. So when tobj is NOT NULL,
3079          * it must be local one.
3080          */
3081         if (tobj && mdd_object_exists(mdd_tobj)) {
3082                 mdd_write_lock(env, mdd_tobj, DT_TGT_CHILD);
3083                 tobj_locked = 1;
3084                 if (mdd_is_dead_obj(mdd_tobj)) {
3085                         /* shld not be dead, something is wrong */
3086                         CERROR("tobj is dead, something is wrong\n");
3087                         rc = -EINVAL;
3088                         goto cleanup;
3089                 }
3090                 mdo_ref_del(env, mdd_tobj, handle);
3091
3092                 /* Remove dot reference. */
3093                 if (S_ISDIR(tattr->la_mode))
3094                         mdo_ref_del(env, mdd_tobj, handle);
3095                 tobj_ref = 1;
3096
3097                 /* fetch updated nlink */
3098                 rc = mdd_la_get(env, mdd_tobj, tattr);
3099                 if (rc != 0) {
3100                         CERROR("%s: Failed to get nlink for tobj "
3101                                 DFID": rc = %d\n",
3102                                 mdd2obd_dev(mdd)->obd_name,
3103                                 PFID(tpobj_fid), rc);
3104                         GOTO(fixup_tpobj, rc);
3105                 }
3106
3107                 la->la_valid = LA_CTIME;
3108                 rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
3109                 if (rc != 0) {
3110                         CERROR("%s: Failed to set ctime for tobj "
3111                                 DFID": rc = %d\n",
3112                                 mdd2obd_dev(mdd)->obd_name,
3113                                 PFID(tpobj_fid), rc);
3114                         GOTO(fixup_tpobj, rc);
3115                 }
3116
3117                 /* XXX: this transfer to ma will be removed with LOD/OSP */
3118                 ma->ma_attr = *tattr;
3119                 ma->ma_valid |= MA_INODE;
3120                 rc = mdd_finish_unlink(env, mdd_tobj, ma, mdd_tpobj, ltname,
3121                                        handle);
3122                 if (rc != 0) {
3123                         CERROR("%s: Failed to unlink tobj "
3124                                 DFID": rc = %d\n",
3125                                 mdd2obd_dev(mdd)->obd_name,
3126                                 PFID(tpobj_fid), rc);
3127                         GOTO(fixup_tpobj, rc);
3128                 }
3129
3130                 /* fetch updated nlink */
3131                 rc = mdd_la_get(env, mdd_tobj, tattr);
3132                 if (rc == -ENOENT) {
3133                         /* the object got removed, let's
3134                          * return the latest known attributes */
3135                         tattr->la_nlink = 0;
3136                         rc = 0;
3137                 } else if (rc != 0) {
3138                         CERROR("%s: Failed to get nlink for tobj "
3139                                 DFID": rc = %d\n",
3140                                 mdd2obd_dev(mdd)->obd_name,
3141                                 PFID(tpobj_fid), rc);
3142                         GOTO(fixup_tpobj, rc);
3143                 }
3144                 /* XXX: this transfer to ma will be removed with LOD/OSP */
3145                 ma->ma_attr = *tattr;
3146                 ma->ma_valid |= MA_INODE;
3147
3148                 if (tattr->la_nlink == 0)
3149                         cl_flags |= CLF_RENAME_LAST;
3150                 else
3151                         cl_flags &= ~CLF_RENAME_LAST_EXISTS;
3152         }
3153
3154         la->la_valid = LA_CTIME | LA_MTIME;
3155         rc = mdd_update_time(env, mdd_spobj, pattr, la, handle);
3156         if (rc)
3157                 GOTO(fixup_tpobj, rc);
3158
3159         if (mdd_spobj != mdd_tpobj) {
3160                 la->la_valid = LA_CTIME | LA_MTIME;
3161                 rc = mdd_update_time(env, mdd_tpobj, tpattr, la, handle);
3162                 if (rc != 0)
3163                         GOTO(fixup_tpobj, rc);
3164         }
3165
3166         EXIT;
3167
3168 fixup_tpobj:
3169         if (rc) {
3170                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
3171                 if (rc2)
3172                         CWARN("tp obj fix error %d\n",rc2);
3173
3174                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
3175                     !mdd_is_dead_obj(mdd_tobj)) {
3176                         if (tobj_ref) {
3177                                 mdo_ref_add(env, mdd_tobj, handle);
3178                                 if (is_dir)
3179                                         mdo_ref_add(env, mdd_tobj, handle);
3180                         }
3181
3182                         rc2 = __mdd_index_insert(env, mdd_tpobj,
3183                                                   mdo2fid(mdd_tobj),
3184                                                   mdd_object_type(mdd_tobj),
3185                                                   tname, handle);
3186                         if (rc2 != 0)
3187                                 CWARN("tp obj fix error: rc = %d\n", rc2);
3188                 }
3189         }
3190
3191 fixup_spobj:
3192         if (rc && is_dir && mdd_sobj && mdd_spobj != mdd_tpobj) {
3193                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3194                 if (rc2)
3195                         CWARN("%s: sp obj dotdot delete error: rc = %d\n",
3196                                mdd2obd_dev(mdd)->obd_name, rc2);
3197
3198
3199                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid, S_IFDIR,
3200                                               dotdot, handle);
3201                 if (rc2 != 0)
3202                         CWARN("%s: sp obj dotdot insert error: rc = %d\n",
3203                               mdd2obd_dev(mdd)->obd_name, rc2);
3204         }
3205
3206 fixup_spobj2:
3207         if (rc != 0) {
3208                 rc2 = __mdd_index_insert(env, mdd_spobj, lf,
3209                                          mdd_object_type(mdd_sobj), sname,
3210                                          handle);
3211                 if (rc2 != 0)
3212                         CWARN("sp obj fix error: rc = %d\n", rc2);
3213         }
3214
3215 cleanup:
3216         if (tobj_locked)
3217                 mdd_write_unlock(env, mdd_tobj);
3218
3219         if (rc == 0)
3220                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, cl_flags,
3221                                             mdd_tobj, tpobj_fid, lf, spobj_fid,
3222                                             ltname, lsname, handle);
3223
3224 stop:
3225         rc = mdd_trans_stop(env, mdd, rc, handle);
3226
3227 out_pending:
3228         mdd_object_put(env, mdd_sobj);
3229         return rc;
3230 }
3231
3232 /**
3233  * Check whether we should migrate the file/dir
3234  * return val
3235  *      < 0  permission check failed or other error.
3236  *      = 0  the file can be migrated.
3237  **/
3238 static int mdd_migrate_sanity_check(const struct lu_env *env,
3239                                     struct mdd_device *mdd,
3240                                     struct mdd_object *spobj,
3241                                     struct mdd_object *tpobj,
3242                                     struct mdd_object *sobj,
3243                                     struct mdd_object *tobj,
3244                                     const struct lu_attr *spattr,
3245                                     const struct lu_attr *tpattr,
3246                                     const struct lu_attr *attr)
3247 {
3248         int rc;
3249
3250         ENTRY;
3251
3252         if (!mdd_object_remote(sobj)) {
3253                 mdd_read_lock(env, sobj, DT_SRC_CHILD);
3254                 if (sobj->mod_count > 0) {
3255                         CDEBUG(D_INFO, "%s: "DFID" is opened, count %d\n",
3256                                mdd2obd_dev(mdd)->obd_name, PFID(mdo2fid(sobj)),
3257                                sobj->mod_count);
3258                         mdd_read_unlock(env, sobj);
3259                         RETURN(-EBUSY);
3260                 }
3261                 mdd_read_unlock(env, sobj);
3262         }
3263
3264         if (mdd_object_exists(tobj))
3265                 RETURN(-EEXIST);
3266
3267         rc = mdd_rename_sanity_check(env, spobj, spattr, tpobj, tpattr, sobj,
3268                                      attr, NULL, NULL);
3269         RETURN(rc);
3270 }
3271
3272 typedef int (*mdd_dir_stripe_cb)(const struct lu_env *env,
3273                                  struct mdd_object *obj,
3274                                  struct mdd_object *stripe,
3275                                  const struct lu_buf *lmv_buf,
3276                                  const struct lu_buf *lmu_buf,
3277                                  int index,
3278                                  struct thandle *handle);
3279
3280 static int mdd_dir_declare_delete_stripe(const struct lu_env *env,
3281                                          struct mdd_object *obj,
3282                                          struct mdd_object *stripe,
3283                                          const struct lu_buf *lmv_buf,
3284                                          const struct lu_buf *lmu_buf,
3285                                          int index,
3286                                          struct thandle *handle)
3287 {
3288         struct mdd_thread_info *info = mdd_env_info(env);
3289         char *stripe_name = info->mti_name;
3290         struct lmv_user_md *lmu = lmu_buf->lb_buf;
3291         int rc;
3292
3293         if (index < le32_to_cpu(lmu->lum_stripe_count))
3294                 return 0;
3295
3296         rc = mdo_declare_index_delete(env, stripe, dotdot, handle);
3297         if (rc)
3298                 return rc;
3299
3300         snprintf(stripe_name, sizeof(info->mti_name), DFID":%d",
3301                  PFID(mdd_object_fid(stripe)), index);
3302
3303         rc = mdo_declare_index_delete(env, obj, stripe_name, handle);
3304         if (rc)
3305                 return rc;
3306
3307         rc = mdo_declare_ref_del(env, obj, handle);
3308
3309         return rc;
3310 }
3311
3312 /* delete stripe from its master object namespace */
3313 static int mdd_dir_delete_stripe(const struct lu_env *env,
3314                                  struct mdd_object *obj,
3315                                  struct mdd_object *stripe,
3316                                  const struct lu_buf *lmv_buf,
3317                                  const struct lu_buf *lmu_buf,
3318                                  int index,
3319                                  struct thandle *handle)
3320 {
3321         struct mdd_thread_info *info = mdd_env_info(env);
3322         char *stripe_name = info->mti_name;
3323         struct lmv_mds_md_v1 *lmv = lmv_buf->lb_buf;
3324         struct lmv_user_md *lmu = lmu_buf->lb_buf;
3325         __u32 del_offset = le32_to_cpu(lmu->lum_stripe_count);
3326         int rc;
3327
3328         ENTRY;
3329
3330         /* local dir will delete via LOD */
3331         LASSERT(mdd_object_remote(obj));
3332         LASSERT(del_offset < le32_to_cpu(lmv->lmv_stripe_count));
3333
3334         if (index < del_offset)
3335                 RETURN(0);
3336
3337         mdd_write_lock(env, stripe, DT_SRC_CHILD);
3338         rc = __mdd_index_delete_only(env, stripe, dotdot, handle);
3339         if (rc)
3340                 GOTO(out, rc);
3341
3342         snprintf(stripe_name, sizeof(info->mti_name), DFID":%d",
3343                  PFID(mdd_object_fid(stripe)), index);
3344
3345         rc = __mdd_index_delete_only(env, obj, stripe_name, handle);
3346         if (rc)
3347                 GOTO(out, rc);
3348
3349         rc = mdo_ref_del(env, obj, handle);
3350         GOTO(out, rc);
3351 out:
3352         mdd_write_unlock(env, stripe);
3353
3354         return rc;
3355 }
3356
3357 static int mdd_dir_declare_destroy_stripe(const struct lu_env *env,
3358                                           struct mdd_object *obj,
3359                                           struct mdd_object *stripe,
3360                                           const struct lu_buf *lmv_buf,
3361                                           const struct lu_buf *lmu_buf,
3362                                           int index,
3363                                           struct thandle *handle)
3364 {
3365         struct lmv_user_md *lmu = lmu_buf->lb_buf;
3366         __u32 shrink_offset = le32_to_cpu(lmu->lum_stripe_count);
3367         int rc;
3368
3369         if (index < shrink_offset) {
3370                 if (shrink_offset < 2)
3371                         return 0;
3372                 return mdo_declare_xattr_set(env, stripe, lmv_buf,
3373                                              XATTR_NAME_LMV".set", 0, handle);
3374         }
3375
3376         rc = mdo_declare_ref_del(env, stripe, handle);
3377         if (rc)
3378                 return rc;
3379
3380         rc = mdo_declare_destroy(env, stripe, handle);
3381
3382         return rc;
3383 }
3384
3385 static int mdd_dir_destroy_stripe(const struct lu_env *env,
3386                                   struct mdd_object *obj,
3387                                   struct mdd_object *stripe,
3388                                   const struct lu_buf *lmv_buf,
3389                                   const struct lu_buf *lmu_buf,
3390                                   int index,
3391                                   struct thandle *handle)
3392 {
3393         struct mdd_thread_info *info = mdd_env_info(env);
3394         struct lmv_mds_md_v1 *lmv = lmv_buf->lb_buf;
3395         struct lmv_user_md *lmu = lmu_buf->lb_buf;
3396         __u32 shrink_offset = le32_to_cpu(lmu->lum_stripe_count);
3397         int rc;
3398
3399         ENTRY;
3400
3401         /* update remaining stripes' LMV */
3402         if (index < shrink_offset) {
3403                 struct lmv_mds_md_v1 *slave_lmv;
3404                 struct lu_buf slave_buf = {
3405                                 .lb_buf = &info->mti_lmv.lmv_md_v1,
3406                                 .lb_len = sizeof(*slave_lmv)
3407                 };
3408                 __u32 version = le32_to_cpu(lmv->lmv_layout_version);
3409
3410                 /* if dir will be shrunk to 1-stripe, don't update */
3411                 if (shrink_offset < 2)
3412                         RETURN(0);
3413
3414                 slave_lmv = slave_buf.lb_buf;
3415                 memset(slave_lmv, 0, sizeof(*slave_lmv));
3416                 slave_lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_STRIPE);
3417                 slave_lmv->lmv_stripe_count = lmu->lum_stripe_count;
3418                 slave_lmv->lmv_master_mdt_index = cpu_to_le32(index);
3419                 slave_lmv->lmv_hash_type = lmv->lmv_hash_type &
3420                                            cpu_to_le32(LMV_HASH_TYPE_MASK);
3421                 slave_lmv->lmv_layout_version = cpu_to_le32(++version);
3422
3423                 rc = mdo_xattr_set(env, stripe, &slave_buf,
3424                                    XATTR_NAME_LMV".set", 0, handle);
3425                 RETURN(rc);
3426         }
3427
3428         mdd_write_lock(env, stripe, DT_SRC_CHILD);
3429         rc = mdo_ref_del(env, stripe, handle);
3430         if (!rc)
3431                 rc = mdo_destroy(env, stripe, handle);
3432         mdd_write_unlock(env, stripe);
3433
3434         RETURN(rc);
3435 }
3436
3437 static int mdd_shrink_stripe_is_empty(const struct lu_env *env,
3438                                        struct mdd_object *obj,
3439                                        struct mdd_object *stripe,
3440                                        const struct lu_buf *lmv_buf,
3441                                        const struct lu_buf *lmu_buf,
3442                                        int index,
3443                                        struct thandle *handle)
3444 {
3445         struct lmv_user_md *lmu = lmu_buf->lb_buf;
3446         __u32 shrink_offset = le32_to_cpu(lmu->lum_stripe_count);
3447
3448         /* the default value is 0, but it means 1 */
3449         if (!shrink_offset)
3450                 shrink_offset = 1;
3451
3452         if (index < shrink_offset)
3453                 return 0;
3454
3455         return mdd_dir_is_empty(env, stripe);
3456 }
3457
3458 /*
3459  * iterate stripes of striped directory on remote MDT, local striped directory
3460  * is accessed via LOD.
3461  */
3462 static int mdd_dir_iterate_stripes(const struct lu_env *env,
3463                                    struct mdd_object *obj,
3464                                    const struct lu_buf *lmv_buf,
3465                                    const struct lu_buf *lmu_buf,
3466                                    struct thandle *handle,
3467                                    mdd_dir_stripe_cb cb)
3468 {
3469         struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
3470         struct lu_fid *fid = &mdd_env_info(env)->mti_fid2;
3471         struct lmv_mds_md_v1 *lmv = lmv_buf->lb_buf;
3472         struct mdd_object *stripe;
3473         int i;
3474         int rc;
3475
3476         ENTRY;
3477
3478         LASSERT(lmv);
3479
3480         for (i = 0; i < le32_to_cpu(lmv->lmv_stripe_count); i++) {
3481                 fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[i]);
3482                 if (!fid_is_sane(fid))
3483                         continue;
3484
3485                 stripe = mdd_object_find(env, mdd, fid);
3486                 if (IS_ERR(stripe))
3487                         RETURN(PTR_ERR(stripe));
3488
3489                 rc = cb(env, obj, stripe, lmv_buf, lmu_buf, i, handle);
3490                 mdd_object_put(env, stripe);
3491                 if (rc)
3492                         RETURN(rc);
3493         }
3494
3495         RETURN(0);
3496 }
3497
3498 typedef int (*mdd_xattr_cb)(const struct lu_env *env,
3499                             struct mdd_object *obj,
3500                             const struct lu_buf *buf,
3501                             const char *name,
3502                             int fl, struct thandle *handle);
3503
3504 /* iterate xattrs, but ignore LMA, LMV, and LINKEA if 'skip_linkea' is set. */
3505 static int mdd_iterate_xattrs(const struct lu_env *env,
3506                               struct mdd_object *sobj,
3507                               struct mdd_object *tobj,
3508                               bool skip_linkea,
3509                               struct thandle *handle,
3510                               mdd_xattr_cb cb)
3511 {
3512         struct mdd_thread_info *info = mdd_env_info(env);
3513         char *xname;
3514         struct lu_buf list_xbuf;
3515         struct lu_buf cbxbuf;
3516         struct lu_buf xbuf = { NULL };
3517         int list_xsize;
3518         int xlen;
3519         int rem;
3520         int xsize;
3521         int rc;
3522
3523         ENTRY;
3524
3525         /* retrieve xattr list from the old object */
3526         list_xsize = mdo_xattr_list(env, sobj, &LU_BUF_NULL);
3527         if (list_xsize == -ENODATA)
3528                 RETURN(0);
3529
3530         if (list_xsize < 0)
3531                 RETURN(list_xsize);
3532
3533         lu_buf_check_and_alloc(&info->mti_big_buf, list_xsize);
3534         if (info->mti_big_buf.lb_buf == NULL)
3535                 RETURN(-ENOMEM);
3536
3537         list_xbuf.lb_buf = info->mti_big_buf.lb_buf;
3538         list_xbuf.lb_len = list_xsize;
3539         rc = mdo_xattr_list(env, sobj, &list_xbuf);
3540         if (rc < 0)
3541                 RETURN(rc);
3542
3543         rem = rc;
3544         rc = 0;
3545         xname = list_xbuf.lb_buf;
3546         while (rem > 0) {
3547                 xlen = strnlen(xname, rem - 1) + 1;
3548                 if (strcmp(XATTR_NAME_LMA, xname) == 0 ||
3549                     strcmp(XATTR_NAME_LMV, xname) == 0)
3550                         goto next;
3551
3552                 if (skip_linkea &&
3553                     strcmp(XATTR_NAME_LINK, xname) == 0)
3554                         goto next;
3555
3556                 xsize = mdo_xattr_get(env, sobj, &LU_BUF_NULL, xname);
3557                 if (xsize == -ENODATA)
3558                         goto next;
3559                 if (xsize < 0)
3560                         GOTO(out, rc = xsize);
3561
3562                 lu_buf_check_and_alloc(&xbuf, xsize);
3563                 if (xbuf.lb_buf == NULL)
3564                         GOTO(out, rc = -ENOMEM);
3565
3566                 rc = mdo_xattr_get(env, sobj, &xbuf, xname);
3567                 if (rc == -ENODATA)
3568                         goto next;
3569                 if (rc < 0)
3570                         GOTO(out, rc);
3571
3572                 cbxbuf = xbuf;
3573 repeat:
3574                 rc = cb(env, tobj, &cbxbuf, xname, 0, handle);
3575                 if (unlikely(rc == -ENOSPC &&
3576                              strcmp(xname, XATTR_NAME_LINK) == 0)) {
3577                         rc = linkea_overflow_shrink(
3578                                         (struct linkea_data *)(cbxbuf.lb_buf));
3579                         if (likely(rc > 0)) {
3580                                 cbxbuf.lb_len = rc;
3581                                 goto repeat;
3582                         }
3583                 }
3584
3585                 if (rc)
3586                         GOTO(out, rc);
3587 next:
3588                 xname += xlen;
3589                 rem -= xlen;
3590         }
3591
3592 out:
3593         lu_buf_free(&xbuf);
3594         RETURN(rc);
3595 }
3596
3597 typedef int (*mdd_linkea_cb)(const struct lu_env *env,
3598                              struct mdd_object *sobj,
3599                              struct mdd_object *tobj,
3600                              const struct lu_name *sname,
3601                              const struct lu_fid *sfid,
3602                              const struct lu_name *lname,
3603                              const struct lu_fid *fid,
3604                              void *opaque,
3605                              struct thandle *handle);
3606
3607 static int mdd_declare_update_link(const struct lu_env *env,
3608                                    struct mdd_object *sobj,
3609                                    struct mdd_object *tobj,
3610                                    const struct lu_name *tname,
3611                                    const struct lu_fid *tpfid,
3612                                    const struct lu_name *lname,
3613                                    const struct lu_fid *fid,
3614                                    void *unused,
3615                                    struct thandle *handle)
3616 {
3617         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3618         struct mdd_object *pobj;
3619         int rc;
3620
3621         /* ignore tobj */
3622         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3623             !strcmp(tname->ln_name, lname->ln_name))
3624                 return 0;
3625
3626         pobj = mdd_object_find(env, mdd, fid);
3627         if (IS_ERR(pobj))
3628                 return PTR_ERR(pobj);
3629
3630
3631         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
3632         if (!rc)
3633                 rc = mdo_declare_index_insert(env, pobj, mdo2fid(tobj),
3634                                               mdd_object_type(sobj),
3635                                               lname->ln_name, handle);
3636         mdd_object_put(env, pobj);
3637         if (rc)
3638                 return rc;
3639
3640         rc = mdo_declare_ref_add(env, tobj, handle);
3641         if (rc)
3642                 return rc;
3643
3644         rc = mdo_declare_ref_del(env, sobj, handle);
3645         return rc;
3646 }
3647
3648 static int mdd_update_link(const struct lu_env *env,
3649                            struct mdd_object *sobj,
3650                            struct mdd_object *tobj,
3651                            const struct lu_name *tname,
3652                            const struct lu_fid *tpfid,
3653                            const struct lu_name *lname,
3654                            const struct lu_fid *fid,
3655                            void *unused,
3656                            struct thandle *handle)
3657 {
3658         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3659         struct mdd_object *pobj;
3660         int rc;
3661
3662         ENTRY;
3663
3664         LASSERT(lu_name_is_valid(lname));
3665
3666         /* ignore tobj */
3667         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3668             !strncmp(tname->ln_name, lname->ln_name, lname->ln_namelen))
3669                 RETURN(0);
3670
3671         CDEBUG(D_INFO, "update "DFID"/"DNAME":"DFID"\n",
3672                PFID(fid), PNAME(lname), PFID(mdo2fid(tobj)));
3673
3674         pobj = mdd_object_find(env, mdd, fid);
3675         if (IS_ERR(pobj)) {
3676                 CWARN("%s: cannot find obj "DFID": %ld\n",
3677                       mdd2obd_dev(mdd)->obd_name, PFID(fid), PTR_ERR(pobj));
3678                 RETURN(PTR_ERR(pobj));
3679         }
3680
3681         if (!mdd_object_exists(pobj)) {
3682                 CDEBUG(D_INFO, DFID" doesn't exist\n", PFID(fid));
3683                 mdd_object_put(env, pobj);
3684                 RETURN(-ENOENT);
3685         }
3686
3687         mdd_write_lock(env, pobj, DT_TGT_PARENT);
3688         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
3689         if (!rc)
3690                 rc = __mdd_index_insert_only(env, pobj, mdo2fid(tobj),
3691                                              mdd_object_type(sobj),
3692                                              lname->ln_name, handle);
3693         mdd_write_unlock(env, pobj);
3694         mdd_object_put(env, pobj);
3695         if (rc)
3696                 RETURN(rc);
3697
3698         mdd_write_lock(env, tobj, DT_TGT_CHILD);
3699         rc = mdo_ref_add(env, tobj, handle);
3700         mdd_write_unlock(env, tobj);
3701         if (rc)
3702                 RETURN(rc);
3703
3704         mdd_write_lock(env, sobj, DT_SRC_CHILD);
3705         rc = mdo_ref_del(env, sobj, handle);
3706         mdd_write_unlock(env, sobj);
3707
3708         RETURN(rc);
3709 }
3710
3711 static inline int mdd_fld_lookup(const struct lu_env *env,
3712                                  struct mdd_device *mdd,
3713                                  const struct lu_fid *fid,
3714                                  __u32 *mdt_index)
3715 {
3716         struct lu_seq_range *range = &mdd_env_info(env)->mti_range;
3717         struct seq_server_site *ss;
3718         int rc;
3719
3720         ss = mdd->mdd_md_dev.md_lu_dev.ld_site->ld_seq_site;
3721
3722         range->lsr_flags = LU_SEQ_RANGE_MDT;
3723         rc = fld_server_lookup(env, ss->ss_server_fld, fid->f_seq, range);
3724         if (rc)
3725                 return rc;
3726
3727         *mdt_index = range->lsr_index;
3728
3729         return 0;
3730 }
3731
3732 static int mdd_is_link_on_source_mdt(const struct lu_env *env,
3733                                      struct mdd_object *sobj,
3734                                      struct mdd_object *tobj,
3735                                      const struct lu_name *tname,
3736                                      const struct lu_fid *tpfid,
3737                                      const struct lu_name *lname,
3738                                      const struct lu_fid *fid,
3739                                      void *opaque,
3740                                      struct thandle *handle)
3741 {
3742         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3743         __u32 source_mdt_index = *(__u32 *)opaque;
3744         __u32 link_mdt_index;
3745         int rc;
3746
3747         ENTRY;
3748
3749         /* ignore tobj */
3750         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3751             !strcmp(tname->ln_name, lname->ln_name))
3752                 return 0;
3753
3754         rc = mdd_fld_lookup(env, mdd, fid, &link_mdt_index);
3755         if (rc)
3756                 RETURN(rc);
3757
3758         RETURN(link_mdt_index == source_mdt_index);
3759 }
3760
3761 static int mdd_iterate_linkea(const struct lu_env *env,
3762                               struct mdd_object *sobj,
3763                               struct mdd_object *tobj,
3764                               const struct lu_name *tname,
3765                               const struct lu_fid *tpfid,
3766                               struct linkea_data *ldata,
3767                               void *opaque,
3768                               struct thandle *handle,
3769                               mdd_linkea_cb cb)
3770 {
3771         struct mdd_thread_info *info = mdd_env_info(env);
3772         char *filename = info->mti_name;
3773         struct lu_name lname;
3774         struct lu_fid fid;
3775         int rc = 0;
3776
3777         if (!ldata->ld_buf)
3778                 return 0;
3779
3780         for (linkea_first_entry(ldata); ldata->ld_lee && !rc;
3781              linkea_next_entry(ldata)) {
3782                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
3783                                     &fid);
3784
3785                 /* Note: lname might miss \0 at the end */
3786                 snprintf(filename, sizeof(info->mti_name), "%.*s",
3787                          lname.ln_namelen, lname.ln_name);
3788                 lname.ln_name = filename;
3789
3790                 CDEBUG(D_INFO, DFID"/"DNAME"\n", PFID(&fid), PNAME(&lname));
3791
3792                 rc = cb(env, sobj, tobj, tname, tpfid, &lname, &fid, opaque,
3793                         handle);
3794         }
3795
3796         return rc;
3797 }
3798
3799 /**
3800  * Prepare linkea, and check whether file needs migrate: if source still has
3801  * link on source MDT, no need to migrate, just update namespace on source and
3802  * target parents.
3803  *
3804  * \retval      0 do migrate
3805  * \retval      1 don't migrate
3806  * \retval      -errno on failure
3807  */
3808 static int migrate_linkea_prepare(const struct lu_env *env,
3809                                   struct mdd_device *mdd,
3810                                   struct mdd_object *spobj,
3811                                   struct mdd_object *tpobj,
3812                                   struct mdd_object *sobj,
3813                                   const struct lu_name *lname,
3814                                   const struct lu_attr *attr,
3815                                   struct linkea_data *ldata)
3816 {
3817         __u32 source_mdt_index;
3818         int rc;
3819
3820         ENTRY;
3821
3822         memset(ldata, 0, sizeof(*ldata));
3823         rc = mdd_linkea_prepare(env, sobj, mdo2fid(spobj), lname,
3824                                 mdo2fid(tpobj), lname, 1, 0, ldata);
3825         if (rc)
3826                 RETURN(rc);
3827
3828         /*
3829          * Then it will check if the file should be migrated. If the file has
3830          * mulitple links, we only need migrate the file if all of its entries
3831          * has been migrated to the remote MDT.
3832          */
3833         if (S_ISDIR(attr->la_mode) || attr->la_nlink < 2)
3834                 RETURN(0);
3835
3836         /* If there are still links locally, don't migrate this file */
3837         LASSERT(ldata->ld_leh != NULL);
3838
3839         /*
3840          * If linkEA is overflow, it means there are some unknown name entries
3841          * under unknown parents, which will prevent the migration.
3842          */
3843         if (unlikely(ldata->ld_leh->leh_overflow_time))
3844                 RETURN(-EOVERFLOW);
3845
3846         rc = mdd_fld_lookup(env, mdd, mdo2fid(sobj), &source_mdt_index);
3847         if (rc)
3848                 RETURN(rc);
3849
3850         rc = mdd_iterate_linkea(env, sobj, NULL, lname, mdo2fid(tpobj), ldata,
3851                                 &source_mdt_index, NULL,
3852                                 mdd_is_link_on_source_mdt);
3853         RETURN(rc);
3854 }
3855
3856 static int mdd_dir_declare_layout_delete(const struct lu_env *env,
3857                                          struct mdd_object *obj,
3858                                          const struct lu_buf *lmv_buf,
3859                                          const struct lu_buf *lmu_buf,
3860                                          struct thandle *handle)
3861 {
3862         int rc;
3863
3864         if (!lmv_buf->lb_buf)
3865                 rc = mdo_declare_index_delete(env, obj, dotdot, handle);
3866         else if (mdd_object_remote(obj))
3867                 rc = mdd_dir_iterate_stripes(env, obj, lmv_buf, lmu_buf, handle,
3868                                              mdd_dir_declare_delete_stripe);
3869         else
3870                 rc = mdo_declare_xattr_set(env, obj, lmu_buf,
3871                                            XATTR_NAME_LMV".del", 0, handle);
3872
3873         return rc;
3874 }
3875
3876 static int mdd_dir_layout_delete(const struct lu_env *env,
3877                                  struct mdd_object *obj,
3878                                  const struct lu_buf *lmv_buf,
3879                                  const struct lu_buf *lmu_buf,
3880                                  struct thandle *handle)
3881 {
3882         int rc;
3883
3884         ENTRY;
3885
3886         mdd_write_lock(env, obj, DT_SRC_PARENT);
3887         if (!lmv_buf->lb_buf)
3888                 /* normal dir */
3889                 rc = __mdd_index_delete_only(env, obj, dotdot, handle);
3890         else if (mdd_object_remote(obj))
3891                 /* striped, but remote */
3892                 rc = mdd_dir_iterate_stripes(env, obj, lmv_buf, lmu_buf, handle,
3893                                              mdd_dir_delete_stripe);
3894         else
3895                 rc = mdo_xattr_set(env, obj, lmu_buf, XATTR_NAME_LMV".del", 0,
3896                                    handle);
3897         mdd_write_unlock(env, obj);
3898
3899         RETURN(rc);
3900 }
3901
3902 static int mdd_declare_migrate_create(const struct lu_env *env,
3903                                       struct mdd_object *tpobj,
3904                                       struct mdd_object *sobj,
3905                                       struct mdd_object *tobj,
3906                                       const struct lu_name *lname,
3907                                       struct lu_attr *attr,
3908                                       struct lu_buf *sbuf,
3909                                       struct linkea_data *ldata,
3910                                       struct md_op_spec *spec,
3911                                       struct dt_allocation_hint *hint,
3912                                       struct thandle *handle)
3913 {
3914         struct mdd_thread_info *info = mdd_env_info(env);
3915         struct lmv_mds_md_v1 *lmv = sbuf->lb_buf;
3916         int rc;
3917
3918         if (S_ISDIR(attr->la_mode)) {
3919                 struct lu_buf lmu_buf = { NULL };
3920
3921                 if (lmv) {
3922                         struct lmv_user_md *lmu = &info->mti_lmv.lmv_user_md;
3923
3924                         lmu->lum_stripe_count = 0;
3925                         lmu_buf.lb_buf = lmu;
3926                         lmu_buf.lb_len = sizeof(*lmu);
3927                 }
3928
3929                 rc = mdd_dir_declare_layout_delete(env, sobj, sbuf, &lmu_buf,
3930                                                    handle);
3931                 if (rc)
3932                         return rc;
3933
3934                 if (lmv) {
3935                         rc = mdo_declare_xattr_del(env, sobj, XATTR_NAME_LMV,
3936                                                    handle);
3937                         if (rc)
3938                                 return rc;
3939                 }
3940         }
3941
3942         rc = mdd_declare_create(env, mdo2mdd(&tpobj->mod_obj), tpobj, tobj,
3943                                 lname, attr, handle, spec, ldata, NULL, NULL,
3944                                 NULL, hint);
3945         if (rc)
3946                 return rc;
3947
3948         if (S_ISDIR(attr->la_mode) && mdd_dir_is_empty(env, sobj) != 0) {
3949                 if (!lmv) {
3950                         /*
3951                          * if sobj is not striped, fake a 1-stripe LMV, which
3952                          * will be used to generate a compound LMV for tobj.
3953                          */
3954                         LASSERT(sizeof(info->mti_key) >
3955                                 lmv_mds_md_size(1, LMV_MAGIC_V1));
3956                         lmv = (typeof(lmv))info->mti_key;
3957                         memset(lmv, 0, sizeof(*lmv));
3958                         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
3959                         lmv->lmv_stripe_count = cpu_to_le32(1);
3960                         fid_le_to_cpu(&lmv->lmv_stripe_fids[0], mdo2fid(sobj));
3961                         sbuf->lb_buf = lmv;
3962                         sbuf->lb_len = lmv_mds_md_size(1, LMV_MAGIC_V1);
3963
3964                         rc = mdo_declare_xattr_set(env, tobj, sbuf,
3965                                                    XATTR_NAME_LMV".add", 0,
3966                                                    handle);
3967                         sbuf->lb_buf = NULL;
3968                         sbuf->lb_len = 0;
3969                 } else {
3970                         rc = mdo_declare_xattr_set(env, tobj, sbuf,
3971                                                    XATTR_NAME_LMV".add", 0,
3972                                                    handle);
3973                 }
3974                 if (rc)
3975                         return rc;
3976         }
3977
3978         /*
3979          * tobj mode will be used in lod_declare_xattr_set(), but it's not
3980          * createb yet, copy from sobj.
3981          */
3982         tobj->mod_obj.mo_lu.lo_header->loh_attr &= ~S_IFMT;
3983         tobj->mod_obj.mo_lu.lo_header->loh_attr |=
3984                 sobj->mod_obj.mo_lu.lo_header->loh_attr & S_IFMT;
3985
3986         rc = mdd_iterate_xattrs(env, sobj, tobj, true, handle,
3987                                 mdo_declare_xattr_set);
3988         if (rc)
3989                 return rc;
3990
3991         if (S_ISREG(attr->la_mode)) {
3992                 struct lu_buf fid_buf;
3993
3994                 handle->th_complex = 1;
3995
3996                 /* target may be remote, update PFID via sobj. */
3997                 fid_buf.lb_buf = (void *)mdo2fid(tobj);
3998                 fid_buf.lb_len = sizeof(struct lu_fid);
3999                 rc = mdo_declare_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID,
4000                                            0, handle);
4001                 if (rc)
4002                         return rc;
4003
4004                 rc = mdo_declare_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4005                 if (rc)
4006                         return rc;
4007         }
4008
4009         if (!S_ISDIR(attr->la_mode)) {
4010                 rc = mdd_iterate_linkea(env, sobj, tobj, lname, mdo2fid(tpobj),
4011                                         ldata, NULL, handle,
4012                                         mdd_declare_update_link);
4013                 if (rc)
4014                         return rc;
4015
4016                 if (lmv) {
4017                         rc = mdo_declare_xattr_del(env, sobj, XATTR_NAME_LMV,
4018                                                    handle);
4019                         if (rc)
4020                                 return rc;
4021                 }
4022         }
4023
4024         return rc;
4025 }
4026
4027 /**
4028  * Create target, migrate xattrs and update links.
4029  *
4030  * Create target according to \a spec, and then migrate xattrs, if it's
4031  * directory, migrate source stripes to target, else update fid to target
4032  * for links.
4033  *
4034  * \param[in] env       execution environment
4035  * \param[in] tpobj     target parent object
4036  * \param[in] sobj      source object
4037  * \param[in] tobj      target object
4038  * \param[in] lname     file name
4039  * \param[in] attr      source attributes
4040  * \param[in] sbuf      source LMV buf
4041  * \param[in] ldata     source linkea
4042  * \param[in] spec      migrate create spec
4043  * \param[in] hint      target creation hint
4044  * \param[in] handle    tranasction handle
4045  *
4046  * \retval      0 on success
4047  * \retval      -errno on failure
4048  **/
4049 static int mdd_migrate_create(const struct lu_env *env,
4050                               struct mdd_object *tpobj,
4051                               struct mdd_object *sobj,
4052                               struct mdd_object *tobj,
4053                               const struct lu_name *lname,
4054                               struct lu_attr *attr,
4055                               const struct lu_buf *sbuf,
4056                               struct linkea_data *ldata,
4057                               struct md_op_spec *spec,
4058                               struct dt_allocation_hint *hint,
4059                               struct thandle *handle)
4060 {
4061         int rc;
4062
4063         ENTRY;
4064
4065         /*
4066          * directory will migrate sobj stripes to tobj:
4067          * 1. delete stripes from sobj.
4068          * 2. add stripes to tobj, see lod_dir_declare_layout_add().
4069          * 3. create/attach stripes for tobj, see lod_xattr_set_lmv().
4070          */
4071         if (S_ISDIR(attr->la_mode)) {
4072                 struct lu_buf lmu_buf = { NULL };
4073
4074                 if (sbuf->lb_buf) {
4075                         struct mdd_thread_info *info = mdd_env_info(env);
4076                         struct lmv_user_md *lmu = &info->mti_lmv.lmv_user_md;
4077
4078                         lmu->lum_stripe_count = 0;
4079                         lmu_buf.lb_buf = lmu;
4080                         lmu_buf.lb_len = sizeof(*lmu);
4081                 }
4082
4083                 rc = mdd_dir_layout_delete(env, sobj, sbuf, &lmu_buf, handle);
4084                 if (rc)
4085                         RETURN(rc);
4086
4087                 /*
4088                  * delete LMV so that later when destroying sobj it won't delete
4089                  * stripes again.
4090                  */
4091                 if (sbuf->lb_buf) {
4092                         mdd_write_lock(env, sobj, DT_SRC_CHILD);
4093                         rc = mdo_xattr_del(env, sobj, XATTR_NAME_LMV, handle);
4094                         mdd_write_unlock(env, sobj);
4095                         if (rc)
4096                                 RETURN(rc);
4097                 }
4098         }
4099
4100         /* don't set nlink from sobj */
4101         attr->la_valid &= ~LA_NLINK;
4102
4103         rc = mdd_create_object(env, tpobj, tobj, attr, spec, NULL, NULL, NULL,
4104                                hint, handle);
4105         if (rc)
4106                 RETURN(rc);
4107
4108         mdd_write_lock(env, tobj, DT_TGT_CHILD);
4109         rc = mdd_iterate_xattrs(env, sobj, tobj, true, handle, mdo_xattr_set);
4110         mdd_write_unlock(env, tobj);
4111         if (rc)
4112                 RETURN(rc);
4113
4114         /* for regular file, update OST objects XATTR_NAME_FID */
4115         if (S_ISREG(attr->la_mode)) {
4116                 struct lu_buf fid_buf;
4117
4118                 /* target may be remote, update PFID via sobj. */
4119                 fid_buf.lb_buf = (void *)mdo2fid(tobj);
4120                 fid_buf.lb_len = sizeof(struct lu_fid);
4121                 rc = mdo_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID, 0,
4122                                    handle);
4123                 if (rc)
4124                         RETURN(rc);
4125
4126                 /* delete LOV to avoid deleting OST objs when destroying sobj */
4127                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4128                 rc = mdo_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4129                 mdd_write_unlock(env, sobj);
4130                 if (rc)
4131                         RETURN(rc);
4132         }
4133
4134         if (!S_ISDIR(attr->la_mode))
4135                 rc = mdd_iterate_linkea(env, sobj, tobj, lname, mdo2fid(tpobj),
4136                                         ldata, NULL, handle, mdd_update_link);
4137
4138         RETURN(rc);
4139 }
4140
4141 static int mdd_declare_migrate_update(const struct lu_env *env,
4142                                       struct mdd_object *spobj,
4143                                       struct mdd_object *tpobj,
4144                                       struct mdd_object *sobj,
4145                                       struct mdd_object *tobj,
4146                                       const struct lu_name *lname,
4147                                       struct lu_attr *attr,
4148                                       struct lu_attr *spattr,
4149                                       struct lu_attr *tpattr,
4150                                       struct linkea_data *ldata,
4151                                       bool do_create,
4152                                       bool do_destroy,
4153                                       struct md_attr *ma,
4154                                       struct thandle *handle)
4155 {
4156         struct mdd_thread_info *info = mdd_env_info(env);
4157         const struct lu_fid *fid = mdo2fid(do_create ? tobj : sobj);
4158         struct lu_attr *la = &info->mti_la_for_fix;
4159         int rc;
4160
4161         rc = mdo_declare_index_delete(env, spobj, lname->ln_name, handle);
4162         if (rc)
4163                 return rc;
4164
4165         if (S_ISDIR(attr->la_mode)) {
4166                 rc = mdo_declare_ref_del(env, spobj, handle);
4167                 if (rc)
4168                         return rc;
4169         }
4170
4171         rc = mdo_declare_index_insert(env, tpobj, fid, mdd_object_type(sobj),
4172                                       lname->ln_name, handle);
4173         if (rc)
4174                 return rc;
4175
4176         rc = mdd_declare_links_add(env, do_create ? tobj : sobj, handle, ldata);
4177         if (rc)
4178                 return rc;
4179
4180         if (S_ISDIR(attr->la_mode)) {
4181                 rc = mdo_declare_ref_add(env, tpobj, handle);
4182                 if (rc)
4183                         return rc;
4184         }
4185
4186         la->la_valid = LA_CTIME | LA_MTIME;
4187         rc = mdo_declare_attr_set(env, spobj, la, handle);
4188         if (rc)
4189                 return rc;
4190
4191         if (tpobj != spobj) {
4192                 rc = mdo_declare_attr_set(env, tpobj, la, handle);
4193                 if (rc)
4194                         return rc;
4195         }
4196
4197         if (do_create && do_destroy) {
4198                 rc = mdo_declare_ref_del(env, sobj, handle);
4199                 if (rc)
4200                         return rc;
4201
4202                 rc = mdo_declare_destroy(env, sobj, handle);
4203                 if (rc)
4204                         return rc;
4205         }
4206
4207         return rc;
4208 }
4209
4210 /**
4211  * migrate dirent from \a spobj to \a tpobj, and destroy \a sobj
4212  **/
4213 static int mdd_migrate_update(const struct lu_env *env,
4214                               struct mdd_object *spobj,
4215                               struct mdd_object *tpobj,
4216                               struct mdd_object *sobj,
4217                               struct mdd_object *tobj,
4218                               const struct lu_name *lname,
4219                               struct lu_attr *attr,
4220                               struct lu_attr *spattr,
4221                               struct lu_attr *tpattr,
4222                               struct linkea_data *ldata,
4223                               bool do_create,
4224                               bool do_destroy,
4225                               struct md_attr *ma,
4226                               struct thandle *handle)
4227 {
4228         struct mdd_thread_info *info = mdd_env_info(env);
4229         const struct lu_fid *fid = mdo2fid(do_create ? tobj : sobj);
4230         struct lu_attr *la = &info->mti_la_for_fix;
4231         int rc;
4232
4233         ENTRY;
4234
4235         CDEBUG(D_INFO, "update %s "DFID"/"DFID" to "DFID"/"DFID"\n",
4236                lname->ln_name, PFID(mdo2fid(spobj)),
4237                PFID(mdo2fid(sobj)), PFID(mdo2fid(tpobj)),
4238                PFID(fid));
4239
4240         rc = __mdd_index_delete(env, spobj, lname->ln_name,
4241                                 S_ISDIR(attr->la_mode), handle);
4242         if (rc)
4243                 RETURN(rc);
4244
4245         rc = __mdd_index_insert(env, tpobj, fid, mdd_object_type(sobj),
4246                                 lname->ln_name, handle);
4247         if (rc)
4248                 RETURN(rc);
4249
4250         rc = mdd_links_write(env, do_create ? tobj : sobj, ldata, handle);
4251         if (rc)
4252                 RETURN(rc);
4253
4254         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
4255         la->la_valid = LA_CTIME | LA_MTIME;
4256         mdd_write_lock(env, spobj, DT_SRC_PARENT);
4257         rc = mdd_update_time(env, spobj, spattr, la, handle);
4258         mdd_write_unlock(env, spobj);
4259         if (rc)
4260                 RETURN(rc);
4261
4262         if (tpobj != spobj) {
4263                 la->la_valid = LA_CTIME | LA_MTIME;
4264                 mdd_write_lock(env, tpobj, DT_TGT_PARENT);
4265                 rc = mdd_update_time(env, tpobj, tpattr, la, handle);
4266                 mdd_write_unlock(env, tpobj);
4267                 if (rc)
4268                         RETURN(rc);
4269         }
4270
4271         /*
4272          * there are three situations we shouldn't destroy source:
4273          * 1. if source is not dir, and it happens to be located on the same MDT
4274          *    as target parent.
4275          * 2. if source is not dir, and has link on the same MDT where source is
4276          *    located.
4277          * 3. if source is dir, and it's a normal, non-empty dir.
4278          *
4279          * the first two situations equals to !do_create, and the 3rd equals to
4280          * !do_destroy, so the below condition is actually
4281          * !(!do_create || !do_destroy).
4282          *
4283          * NB, if user has opened source dir before migration, he will get
4284          * -ENOENT error when close it later, because source is likely to be
4285          *  remote, which can't be moved to orphan list, but except this error
4286          *  message, this won't cause any inconsistency or trouble.
4287          */
4288         if (do_create && do_destroy) {
4289                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4290                 mdo_ref_del(env, sobj, handle);
4291                 rc = mdo_destroy(env, sobj, handle);
4292                 mdd_write_unlock(env, sobj);
4293         }
4294
4295         RETURN(rc);
4296 }
4297
4298 /**
4299  * Migrate directory or file.
4300  *
4301  * migrate source to target in following steps:
4302  *   1. create target, append source stripes after target's if it's directory,
4303  *      migrate xattrs and update fid of source links.
4304  *   2. update namespace: migrate dirent from source parent to target parent,
4305  *      update file linkea, and destroy source if it's not needed any more.
4306  *
4307  * \param[in] env       execution environment
4308  * \param[in] md_pobj   parent master object
4309  * \param[in] md_sobj   source object
4310  * \param[in] lname     file name
4311  * \param[in] md_tobj   target object
4312  * \param[in] spec      target creation spec
4313  * \param[in] ma        used to update \a pobj mtime and ctime
4314  *
4315  * \retval              0 on success
4316  * \retval              -errno on failure
4317  */
4318 static int mdd_migrate(const struct lu_env *env, struct md_object *md_pobj,
4319                        struct md_object *md_sobj, const struct lu_name *lname,
4320                        struct md_object *md_tobj, struct md_op_spec *spec,
4321                        struct md_attr *ma)
4322 {
4323         struct mdd_device *mdd = mdo2mdd(md_pobj);
4324         struct mdd_thread_info *info = mdd_env_info(env);
4325         struct mdd_object *pobj = md2mdd_obj(md_pobj);
4326         struct mdd_object *sobj = md2mdd_obj(md_sobj);
4327         struct mdd_object *tobj = md2mdd_obj(md_tobj);
4328         struct mdd_object *spobj = NULL;
4329         struct mdd_object *tpobj = NULL;
4330         struct lu_attr *spattr = &info->mti_pattr;
4331         struct lu_attr *tpattr = &info->mti_tpattr;
4332         struct lu_attr *attr = &info->mti_cattr;
4333         struct linkea_data *ldata = &info->mti_link_data;
4334         struct dt_allocation_hint *hint = &info->mti_hint;
4335         struct lu_fid *fid = &info->mti_fid2;
4336         struct lu_buf pbuf = { NULL };
4337         struct lu_buf sbuf = { NULL };
4338         struct lmv_mds_md_v1 *plmv;
4339         struct thandle *handle;
4340         bool do_create = true;
4341         bool do_destroy = true;
4342         int rc;
4343         ENTRY;
4344
4345         rc = mdd_la_get(env, sobj, attr);
4346         if (rc)
4347                 RETURN(rc);
4348
4349         /* locate source and target stripe on pobj, which are the real parent */
4350         rc = mdd_stripe_get(env, pobj, &pbuf, XATTR_NAME_LMV);
4351         if (rc < 0 && rc != -ENODATA)
4352                 RETURN(rc);
4353
4354         plmv = pbuf.lb_buf;
4355         if (plmv) {
4356                 __u32 hash_type = le32_to_cpu(plmv->lmv_hash_type);
4357                 __u32 count = le32_to_cpu(plmv->lmv_stripe_count);
4358                 int index;
4359
4360                 /* locate target parent stripe */
4361                 if (hash_type & LMV_HASH_FLAG_MIGRATION) {
4362                         /*
4363                          * fail check here to make sure top dir migration
4364                          * succeed.
4365                          */
4366                         if (OBD_FAIL_CHECK_RESET(OBD_FAIL_MIGRATE_ENTRIES, 0))
4367                                 GOTO(out, rc = -EIO);
4368                         hash_type &= ~LMV_HASH_FLAG_MIGRATION;
4369                         count = le32_to_cpu(plmv->lmv_migrate_offset);
4370                 }
4371                 index = lmv_name_to_stripe_index(hash_type, count,
4372                                                  lname->ln_name,
4373                                                  lname->ln_namelen);
4374                 if (index < 0)
4375                         GOTO(out, rc = index);
4376
4377                 fid_le_to_cpu(fid, &plmv->lmv_stripe_fids[index]);
4378                 tpobj = mdd_object_find(env, mdd, fid);
4379                 if (IS_ERR(tpobj))
4380                         GOTO(out, rc = PTR_ERR(tpobj));
4381
4382                 /* locate source parent stripe */
4383                 if (le32_to_cpu(plmv->lmv_hash_type) &
4384                     LMV_HASH_FLAG_MIGRATION) {
4385                         hash_type = le32_to_cpu(plmv->lmv_migrate_hash);
4386                         count = le32_to_cpu(plmv->lmv_stripe_count) -
4387                                 le32_to_cpu(plmv->lmv_migrate_offset);
4388
4389                         index = lmv_name_to_stripe_index(hash_type, count,
4390                                                          lname->ln_name,
4391                                                          lname->ln_namelen);
4392                         if (index < 0) {
4393                                 mdd_object_put(env, tpobj);
4394                                 GOTO(out, rc = index);
4395                         }
4396
4397                         index += le32_to_cpu(plmv->lmv_migrate_offset);
4398                         fid_le_to_cpu(fid, &plmv->lmv_stripe_fids[index]);
4399                         spobj = mdd_object_find(env, mdd, fid);
4400                         if (IS_ERR(spobj)) {
4401                                 mdd_object_put(env, tpobj);
4402                                 GOTO(out, rc = PTR_ERR(spobj));
4403                         }
4404                 } else {
4405                         spobj = tpobj;
4406                         mdd_object_get(spobj);
4407                 }
4408         } else {
4409                 tpobj = pobj;
4410                 spobj = pobj;
4411                 mdd_object_get(tpobj);
4412                 mdd_object_get(spobj);
4413         }
4414
4415         rc = mdd_la_get(env, spobj, spattr);
4416         if (rc)
4417                 GOTO(out, rc);
4418
4419         rc = mdd_la_get(env, tpobj, tpattr);
4420         if (rc)
4421                 GOTO(out, rc);
4422
4423         if (S_ISDIR(attr->la_mode)) {
4424                 struct lmv_user_md_v1 *lmu = spec->u.sp_ea.eadata;
4425
4426                 LASSERT(lmu);
4427
4428                 /*
4429                  * if user use default value '0' for stripe_count, we need to
4430                  * adjust it to '1' to create a 1-stripe directory.
4431                  */
4432                 if (lmu->lum_stripe_count == 0) {
4433                         /* eadata is from request, don't alter it */
4434                         info->mti_lmu = *lmu;
4435                         info->mti_lmu.lum_stripe_count = cpu_to_le32(1);
4436                         spec->u.sp_ea.eadata = &info->mti_lmu;
4437                         lmu = spec->u.sp_ea.eadata;
4438                 }
4439
4440                 rc = mdd_stripe_get(env, sobj, &sbuf, XATTR_NAME_LMV);
4441                 if (rc == -ENODATA) {
4442                         if (mdd_dir_is_empty(env, sobj) == 0) {
4443                                 /*
4444                                  * if sobj is empty, and target is not striped,
4445                                  * create target as a normal directory.
4446                                  */
4447                                 if (le32_to_cpu(lmu->lum_stripe_count) == 1) {
4448                                         info->mti_lmu = *lmu;
4449                                         info->mti_lmu.lum_stripe_count = 0;
4450                                         spec->u.sp_ea.eadata = &info->mti_lmu;
4451                                         lmu = spec->u.sp_ea.eadata;
4452                                 }
4453                         } else {
4454                                 /*
4455                                  * sobj is not striped dir, if it's not empty,
4456                                  * it will be migrated to be a stripe of target,
4457                                  * don't destroy it after migration.
4458                                  */
4459                                 do_destroy = false;
4460                         }
4461                 } else if (rc) {
4462                         GOTO(out, rc);
4463                 } else {
4464                         struct lmv_mds_md_v1 *lmv = sbuf.lb_buf;
4465
4466                         if (le32_to_cpu(lmv->lmv_hash_type) &
4467                             LMV_HASH_FLAG_MIGRATION) {
4468                                 __u32 lum_stripe_count = lmu->lum_stripe_count;
4469                                 __u32 lmv_hash_type = lmv->lmv_hash_type &
4470                                         cpu_to_le32(LMV_HASH_TYPE_MASK);
4471
4472                                 if (!lum_stripe_count)
4473                                         lum_stripe_count = cpu_to_le32(1);
4474
4475                                 /* TODO: check specific MDTs */
4476                                 if (lmv->lmv_migrate_offset !=
4477                                     lum_stripe_count ||
4478                                     lmv->lmv_master_mdt_index !=
4479                                     lmu->lum_stripe_offset ||
4480                                     (lmv_hash_type != 0 &&
4481                                      lmv_hash_type != lmu->lum_hash_type)) {
4482                                         CERROR("%s: \'"DNAME"\' migration was "
4483                                                 "interrupted, run \'lfs migrate "
4484                                                 "-m %d -c %d -H %d "DNAME"\' to "
4485                                                 "finish migration.\n",
4486                                                 mdd2obd_dev(mdd)->obd_name,
4487                                                 PNAME(lname),
4488                                                 le32_to_cpu(
4489                                                     lmv->lmv_master_mdt_index),
4490                                                 le32_to_cpu(
4491                                                     lmv->lmv_migrate_offset),
4492                                                 le32_to_cpu(lmv_hash_type),
4493                                                 PNAME(lname));
4494                                         GOTO(out, rc = -EPERM);
4495                                 }
4496                                 GOTO(out, rc = -EALREADY);
4497                         }
4498                 }
4499         } else if (!mdd_object_remote(tpobj)) {
4500                 /*
4501                  * if source is already on MDT where target parent is located,
4502                  * no need to create, just update namespace.
4503                  */
4504                 do_create = false;
4505         } else if (S_ISLNK(attr->la_mode)) {
4506                 lu_buf_check_and_alloc(&sbuf, attr->la_size + 1);
4507                 if (!sbuf.lb_buf)
4508                         GOTO(out, rc = -ENOMEM);
4509                 rc = mdd_readlink(env, &sobj->mod_obj, &sbuf);
4510                 if (rc <= 0) {
4511                         rc = rc ?: -EFAULT;
4512                         CERROR("%s: "DFID" readlink failed: rc = %d\n",
4513                                mdd2obd_dev(mdd)->obd_name,
4514                                PFID(mdo2fid(sobj)), rc);
4515                         GOTO(out, rc);
4516                 }
4517                 spec->u.sp_symname = sbuf.lb_buf;
4518         } else if (S_ISREG(attr->la_mode)) {
4519                 spec->sp_cr_flags |= MDS_OPEN_DELAY_CREATE;
4520                 spec->sp_cr_flags &= ~MDS_OPEN_HAS_EA;
4521         }
4522
4523         /*
4524          * if sobj has link on the same MDT, no need to create, just update
4525          * namespace, and it will be a remote file on target parent, which is
4526          * similar to rename.
4527          */
4528         rc = migrate_linkea_prepare(env, mdd, spobj, tpobj, sobj, lname, attr,
4529                                     ldata);
4530         if (rc > 0)
4531                 do_create = false;
4532         else if (rc)
4533                 GOTO(out, rc);
4534
4535         rc = mdd_migrate_sanity_check(env, mdd, spobj, tpobj, sobj, tobj,
4536                                       spattr, tpattr, attr);
4537         if (rc)
4538                 GOTO(out, rc);
4539
4540         mdd_object_make_hint(env, tpobj, tobj, attr, spec, hint);
4541
4542         handle = mdd_trans_create(env, mdd);
4543         if (IS_ERR(handle))
4544                 GOTO(out, rc = PTR_ERR(handle));
4545
4546         if (do_create) {
4547                 rc = mdd_declare_migrate_create(env, tpobj, sobj, tobj, lname,
4548                                                 attr, &sbuf, ldata, spec, hint,
4549                                                 handle);
4550                 if (rc)
4551                         GOTO(stop_trans, rc);
4552         }
4553
4554         rc = mdd_declare_migrate_update(env, spobj, tpobj, sobj, tobj, lname,
4555                                         attr, spattr, tpattr, ldata, do_create,
4556                                         do_destroy, ma, handle);
4557         if (rc)
4558                 GOTO(stop_trans, rc);
4559
4560         rc = mdd_declare_changelog_store(env, mdd, CL_MIGRATE, lname, NULL,
4561                                          handle);
4562         if (rc)
4563                 GOTO(stop_trans, rc);
4564
4565         rc = mdd_trans_start(env, mdd, handle);
4566         if (rc)
4567                 GOTO(stop_trans, rc);
4568
4569         if (do_create) {
4570                 rc = mdd_migrate_create(env, tpobj, sobj, tobj, lname, attr,
4571                                         &sbuf, ldata, spec, hint, handle);
4572                 if (rc)
4573                         GOTO(stop_trans, rc);
4574         }
4575
4576         rc = mdd_migrate_update(env, spobj, tpobj, sobj, tobj, lname, attr,
4577                                 spattr, tpattr, ldata, do_create, do_destroy,
4578                                 ma, handle);
4579         if (rc)
4580                 GOTO(stop_trans, rc);
4581
4582         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0, tobj,
4583                                     mdo2fid(spobj), mdo2fid(sobj),
4584                                     mdo2fid(tpobj), lname, lname, handle);
4585         if (rc)
4586                 GOTO(stop_trans, rc);
4587
4588         EXIT;
4589 stop_trans:
4590         rc = mdd_trans_stop(env, mdd, rc, handle);
4591 out:
4592         if (spobj && !IS_ERR(spobj))
4593                 mdd_object_put(env, spobj);
4594         if (tpobj && !IS_ERR(tpobj))
4595                 mdd_object_put(env, tpobj);
4596         lu_buf_free(&sbuf);
4597         lu_buf_free(&pbuf);
4598         return rc;
4599 }
4600
4601 static int __mdd_dir_declare_layout_shrink(const struct lu_env *env,
4602                                            struct mdd_object *pobj,
4603                                            struct mdd_object *obj,
4604                                            struct mdd_object *stripe,
4605                                            struct lu_attr *attr,
4606                                            struct lu_buf *lmv_buf,
4607                                            const struct lu_buf *lmu_buf,
4608                                            struct lu_name *lname,
4609                                            struct thandle *handle)
4610 {
4611         struct mdd_thread_info *info = mdd_env_info(env);
4612         struct lmv_mds_md_v1 *lmv = lmv_buf->lb_buf;
4613         struct lmv_user_md *lmu = (typeof(lmu))info->mti_key;
4614         struct lu_buf shrink_buf = { .lb_buf = lmu,
4615                                      .lb_len = sizeof(*lmu) };
4616         int rc;
4617
4618         LASSERT(lmv);
4619
4620         memcpy(lmu, lmu_buf->lb_buf, sizeof(*lmu));
4621
4622         if (le32_to_cpu(lmu->lum_stripe_count) < 2)
4623                 lmu->lum_stripe_count = 0;
4624
4625         rc = mdd_dir_declare_layout_delete(env, obj, lmv_buf, &shrink_buf,
4626                                            handle);
4627         if (rc)
4628                 return rc;
4629
4630         if (lmu->lum_stripe_count == 0) {
4631                 lmu->lum_stripe_count = cpu_to_le32(1);
4632
4633                 rc = mdo_declare_xattr_del(env, obj, XATTR_NAME_LMV, handle);
4634                 if (rc)
4635                         return rc;
4636         }
4637
4638         rc = mdd_dir_iterate_stripes(env, obj, lmv_buf, &shrink_buf, handle,
4639                                      mdd_dir_declare_destroy_stripe);
4640         if (rc)
4641                 return rc;
4642
4643         if (le32_to_cpu(lmu->lum_stripe_count) > 1)
4644                 return mdo_declare_xattr_set(env, obj, lmv_buf,
4645                                              XATTR_NAME_LMV".set", 0, handle);
4646
4647         rc = mdo_declare_index_insert(env, stripe, mdo2fid(pobj), S_IFDIR,
4648                                       dotdot, handle);
4649         if (rc)
4650                 return rc;
4651
4652         rc = mdd_iterate_xattrs(env, obj, stripe, false, handle,
4653                                 mdo_declare_xattr_set);
4654         if (rc)
4655                 return rc;
4656
4657         rc = mdo_declare_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4658         if (rc)
4659                 return rc;
4660
4661         rc = mdo_declare_attr_set(env, stripe, attr, handle);
4662         if (rc)
4663                 return rc;
4664
4665         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
4666         if (rc)
4667                 return rc;
4668
4669         rc = mdo_declare_index_insert(env, pobj, mdo2fid(stripe), attr->la_mode,
4670                                       lname->ln_name, handle);
4671         if (rc)
4672                 return rc;
4673
4674         rc = mdo_declare_ref_del(env, obj, handle);
4675         if (rc)
4676                 return rc;
4677
4678         rc = mdo_declare_ref_del(env, obj, handle);
4679         if (rc)
4680                 return rc;
4681
4682         rc = mdo_declare_destroy(env, obj, handle);
4683         if (rc)
4684                 return rc;
4685
4686         return rc;
4687
4688 }
4689
4690 /*
4691  * after files under \a obj were migrated, shrink old stripes from \a obj,
4692  * furthermore, if it becomes a 1-stripe directory, convert it to a normal one.
4693  */
4694 static int __mdd_dir_layout_shrink(const struct lu_env *env,
4695                                    struct mdd_object *pobj,
4696                                    struct mdd_object *obj,
4697                                    struct mdd_object *stripe,
4698                                    struct lu_attr *attr,
4699                                    struct lu_buf *lmv_buf,
4700                                    const struct lu_buf *lmu_buf,
4701                                    struct lu_name *lname,
4702                                    struct thandle *handle)
4703 {
4704         struct mdd_thread_info *info = mdd_env_info(env);
4705         struct lmv_mds_md_v1 *lmv = lmv_buf->lb_buf;
4706         struct lmv_user_md *lmu = (typeof(lmu))info->mti_key;
4707         struct lu_buf shrink_buf = { .lb_buf = lmu,
4708                                      .lb_len = sizeof(*lmu) };
4709         int len = lmv_buf->lb_len;
4710         __u32 version = le32_to_cpu(lmv->lmv_layout_version);
4711         int rc;
4712
4713         ENTRY;
4714
4715         /* lmu needs to be altered, but lmu_buf is const */
4716         memcpy(lmu, lmu_buf->lb_buf, sizeof(*lmu));
4717
4718         /*
4719          * if dir will be shrunk to 1-stripe, delete all stripes, because it
4720          * will be converted to normal dir.
4721          */
4722         if (le32_to_cpu(lmu->lum_stripe_count) == 1)
4723                 lmu->lum_stripe_count = 0;
4724
4725         /* delete stripes after lmu_stripe_count */
4726         rc = mdd_dir_layout_delete(env, obj, lmv_buf, &shrink_buf, handle);
4727         if (rc)
4728                 RETURN(rc);
4729
4730         if (lmu->lum_stripe_count == 0) {
4731                 lmu->lum_stripe_count = cpu_to_le32(1);
4732
4733                 /* delete LMV to avoid deleting stripes again upon destroy */
4734                 mdd_write_lock(env, obj, DT_SRC_CHILD);
4735                 rc = mdo_xattr_del(env, obj, XATTR_NAME_LMV, handle);
4736                 mdd_write_unlock(env, obj);
4737                 if (rc)
4738                         RETURN(rc);
4739         }
4740
4741         /* destroy stripes after lmu_stripe_count */
4742         mdd_write_lock(env, obj, DT_SRC_PARENT);
4743         rc = mdd_dir_iterate_stripes(env, obj, lmv_buf, &shrink_buf, handle,
4744                                      mdd_dir_destroy_stripe);
4745         mdd_write_unlock(env, obj);
4746
4747         if (le32_to_cpu(lmu->lum_stripe_count) > 1) {
4748                 /* update dir LMV, that's all if it's still striped. */
4749                 lmv->lmv_stripe_count = lmu->lum_stripe_count;
4750                 lmv->lmv_hash_type &= ~cpu_to_le32(LMV_HASH_FLAG_MIGRATION);
4751                 lmv->lmv_migrate_offset = 0;
4752                 lmv->lmv_migrate_hash = 0;
4753                 lmv->lmv_layout_version = cpu_to_le32(++version);
4754
4755                 lmv_buf->lb_len = sizeof(*lmv);
4756                 rc = mdo_xattr_set(env, obj, lmv_buf, XATTR_NAME_LMV".set", 0,
4757                                    handle);
4758                 lmv_buf->lb_len = len;
4759                 RETURN(rc);
4760         }
4761
4762         /* replace directory with its remaining stripe */
4763         LASSERT(pobj);
4764         LASSERT(stripe);
4765
4766         mdd_write_lock(env, pobj, DT_SRC_PARENT);
4767         mdd_write_lock(env, obj, DT_SRC_CHILD);
4768
4769         /* insert dotdot to stripe which points to parent */
4770         rc = __mdd_index_insert_only(env, stripe, mdo2fid(pobj), S_IFDIR,
4771                                      dotdot, handle);
4772         if (rc)
4773                 GOTO(out, rc);
4774
4775         /* copy xattrs including linkea */
4776         rc = mdd_iterate_xattrs(env, obj, stripe, false, handle, mdo_xattr_set);
4777         if (rc)
4778                 GOTO(out, rc);
4779
4780         /* delete LMV */
4781         rc = mdo_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4782         if (rc)
4783                 GOTO(out, rc);
4784
4785         /* don't set nlink from parent */
4786         attr->la_valid &= ~LA_NLINK;
4787
4788         rc = mdo_attr_set(env, stripe, attr, handle);
4789         if (rc)
4790                 GOTO(out, rc);
4791
4792         /* delete dir name from parent */
4793         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
4794         if (rc)
4795                 GOTO(out, rc);
4796
4797         /* insert stripe to parent with dir name */
4798         rc = __mdd_index_insert_only(env, pobj, mdo2fid(stripe), attr->la_mode,
4799                                      lname->ln_name, handle);
4800         if (rc)
4801                 GOTO(out, rc);
4802
4803         /* destroy dir obj */
4804         rc = mdo_ref_del(env, obj, handle);
4805         if (rc)
4806                 GOTO(out, rc);
4807
4808         rc = mdo_ref_del(env, obj, handle);
4809         if (rc)
4810                 GOTO(out, rc);
4811
4812         rc = mdo_destroy(env, obj, handle);
4813         if (rc)
4814                 GOTO(out, rc);
4815
4816         EXIT;
4817 out:
4818         mdd_write_unlock(env, obj);
4819         mdd_write_unlock(env, pobj);
4820
4821         return rc;
4822 }
4823
4824 /*
4825  * shrink directory stripes to lum_stripe_count specified by lum_mds_md.
4826  */
4827 int mdd_dir_layout_shrink(const struct lu_env *env,
4828                           struct md_object *md_obj,
4829                           const struct lu_buf *lmu_buf)
4830 {
4831         struct mdd_device *mdd = mdo2mdd(md_obj);
4832         struct mdd_thread_info *info = mdd_env_info(env);
4833         struct mdd_object *obj = md2mdd_obj(md_obj);
4834         struct mdd_object *pobj = NULL;
4835         struct mdd_object *stripe = NULL;
4836         struct lu_attr *attr = &info->mti_pattr;
4837         struct lu_fid *fid = &info->mti_fid2;
4838         struct lu_name lname = { NULL };
4839         struct lu_buf lmv_buf = { NULL };
4840         struct lmv_mds_md_v1 *lmv;
4841         struct lmv_user_md *lmu;
4842         struct thandle *handle;
4843         int rc;
4844
4845         ENTRY;
4846
4847         rc = mdd_la_get(env, obj, attr);
4848         if (rc)
4849                 RETURN(rc);
4850
4851         if (!S_ISDIR(attr->la_mode))
4852                 RETURN(-ENOTDIR);
4853
4854         rc = mdd_stripe_get(env, obj, &lmv_buf, XATTR_NAME_LMV);
4855         if (rc < 0)
4856                 RETURN(rc);
4857
4858         lmv = lmv_buf.lb_buf;
4859         lmu = lmu_buf->lb_buf;
4860
4861         /* this was checked in MDT */
4862         LASSERT(le32_to_cpu(lmu->lum_stripe_count) <
4863                 le32_to_cpu(lmv->lmv_stripe_count));
4864
4865         rc = mdd_dir_iterate_stripes(env, obj, &lmv_buf, lmu_buf, NULL,
4866                                      mdd_shrink_stripe_is_empty);
4867         if (rc < 0)
4868                 GOTO(out, rc);
4869         else if (rc != 0)
4870                 GOTO(out, rc = -ENOTEMPTY);
4871
4872         /*
4873          * if obj stripe count will be shrunk to 1, we need to convert it to a
4874          * normal dir, which will change its fid and update parent namespace,
4875          * get obj name and parent fid from linkea.
4876          */
4877         if (le32_to_cpu(lmu->lum_stripe_count) < 2) {
4878                 struct linkea_data *ldata = &info->mti_link_data;
4879                 char *filename = info->mti_name;
4880
4881                 rc = mdd_links_read(env, obj, ldata);
4882                 if (rc)
4883                         GOTO(out, rc);
4884
4885                 if (ldata->ld_leh->leh_reccount > 1)
4886                         GOTO(out, rc = -EINVAL);
4887
4888                 linkea_first_entry(ldata);
4889                 if (!ldata->ld_lee)
4890                         GOTO(out, rc = -ENODATA);
4891
4892                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
4893                                     fid);
4894
4895                 /* Note: lname might miss \0 at the end */
4896                 snprintf(filename, sizeof(info->mti_name), "%.*s",
4897                          lname.ln_namelen, lname.ln_name);
4898                 lname.ln_name = filename;
4899
4900                 pobj = mdd_object_find(env, mdd, fid);
4901                 if (IS_ERR(pobj)) {
4902                         rc = PTR_ERR(pobj);
4903                         pobj = NULL;
4904                         GOTO(out, rc);
4905                 }
4906
4907                 fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[0]);
4908
4909                 stripe = mdd_object_find(env, mdd, fid);
4910                 if (IS_ERR(stripe)) {
4911                         mdd_object_put(env, pobj);
4912                         pobj = NULL;
4913                         GOTO(out, rc = PTR_ERR(stripe));
4914                 }
4915         }
4916
4917         handle = mdd_trans_create(env, mdd);
4918         if (IS_ERR(handle))
4919                 GOTO(out, rc = PTR_ERR(handle));
4920
4921         rc = __mdd_dir_declare_layout_shrink(env, pobj, obj, stripe, attr,
4922                                              &lmv_buf, lmu_buf, &lname, handle);
4923         if (rc)
4924                 GOTO(stop_trans, rc);
4925
4926         rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL, NULL,
4927                                          handle);
4928         if (rc)
4929                 GOTO(stop_trans, rc);
4930
4931         rc = mdd_trans_start(env, mdd, handle);
4932         if (rc)
4933                 GOTO(stop_trans, rc);
4934
4935         rc = __mdd_dir_layout_shrink(env, pobj, obj, stripe, attr, &lmv_buf,
4936                                      lmu_buf, &lname, handle);
4937         if (rc)
4938                 GOTO(stop_trans, rc);
4939
4940         rc = mdd_changelog_data_store_xattr(env, mdd, CL_LAYOUT, 0, obj,
4941                                             XATTR_NAME_LMV, handle);
4942         GOTO(stop_trans, rc);
4943
4944 stop_trans:
4945         rc = mdd_trans_stop(env, mdd, rc, handle);
4946 out:
4947         if (pobj) {
4948                 mdd_object_put(env, stripe);
4949                 mdd_object_put(env, pobj);
4950         }
4951         lu_buf_free(&lmv_buf);
4952         return rc;
4953 }
4954
4955 const struct md_dir_operations mdd_dir_ops = {
4956         .mdo_is_subdir     = mdd_is_subdir,
4957         .mdo_lookup        = mdd_lookup,
4958         .mdo_create        = mdd_create,
4959         .mdo_rename        = mdd_rename,
4960         .mdo_link          = mdd_link,
4961         .mdo_unlink        = mdd_unlink,
4962         .mdo_create_data   = mdd_create_data,
4963         .mdo_migrate       = mdd_migrate,
4964 };