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