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