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         if (S_ISDIR(attr->la_mode)) {
2781                 struct lmv_user_md *lmu = spec->u.sp_ea.eadata;
2782
2783                 /*
2784                  * migrate may create 1-stripe directory, so lod_ah_init()
2785                  * doesn't adjust stripe count from lmu.
2786                  */
2787                 if (lmu && lmu->lum_stripe_count == cpu_to_le32(1))
2788                         lmu->lum_stripe_count = 0;
2789         }
2790
2791         mdd_object_make_hint(env, mdd_pobj, son, attr, spec, hint);
2792
2793         memset(ldata, 0, sizeof(*ldata));
2794         if (CFS_FAIL_CHECK(OBD_FAIL_LFSCK_BAD_PARENT)) {
2795                 struct lu_fid tfid = *mdd_object_fid(mdd_pobj);
2796
2797                 tfid.f_oid--;
2798                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2799                                         &tfid, lname, 1, 0, ldata);
2800         } else {
2801                 rc = mdd_linkea_prepare(env, son, NULL, NULL,
2802                                         mdd_object_fid(mdd_pobj),
2803                                         lname, 1, 0, ldata);
2804         }
2805
2806         if (spec->sp_cr_flags & MDS_OPEN_PCC) {
2807                 LASSERT(spec->sp_cr_flags & MDS_OPEN_HAS_EA);
2808
2809                 memset(&hsm_buf, 0, sizeof(hsm_buf));
2810                 lu_buf_alloc(&hsm_buf, sizeof(struct hsm_attrs));
2811                 if (hsm_buf.lb_buf == NULL)
2812                         GOTO(out_stop, rc = -ENOMEM);
2813         }
2814
2815         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, attr,
2816                                 handle, spec, ldata, &def_acl_buf, &acl_buf,
2817                                 &hsm_buf, hint);
2818         if (rc)
2819                 GOTO(out_stop, rc);
2820
2821         rc = mdd_trans_start(env, mdd, handle);
2822         if (rc)
2823                 GOTO(out_stop, rc);
2824
2825         rc = mdd_create_object(env, mdd_pobj, son, attr, spec, &acl_buf,
2826                                &def_acl_buf, &hsm_buf, hint, handle, true);
2827         if (rc != 0)
2828                 GOTO(out_stop, rc);
2829
2830         if (unlikely(spec->sp_cr_flags & MDS_OPEN_VOLATILE)) {
2831                 mdd_write_lock(env, son, DT_TGT_CHILD);
2832                 son->mod_flags |= VOLATILE_OBJ;
2833                 rc = mdd_orphan_insert(env, son, handle);
2834                 GOTO(out_volatile, rc);
2835         } else {
2836                 rc = __mdd_index_insert(env, mdd_pobj, mdd_object_fid(son),
2837                                         attr->la_mode, name, handle);
2838                 if (rc != 0)
2839                         GOTO(err_created, rc);
2840
2841                 mdd_links_add(env, son, mdd_object_fid(mdd_pobj), lname,
2842                               handle, ldata, 1);
2843
2844                 /* update parent directory mtime/ctime */
2845                 *la = *attr;
2846                 la->la_valid = LA_CTIME | LA_MTIME;
2847                 rc = mdd_update_time(env, mdd_pobj, pattr, la, handle);
2848                 if (rc)
2849                         GOTO(err_insert, rc);
2850         }
2851
2852         EXIT;
2853 err_insert:
2854         if (rc != 0) {
2855                 if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
2856                         rc2 = mdd_orphan_delete(env, son, handle);
2857                 else
2858                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2859                                                  S_ISDIR(attr->la_mode),
2860                                                  handle);
2861                 if (rc2 != 0)
2862                         goto out_stop;
2863
2864 err_created:
2865                 mdd_write_lock(env, son, DT_TGT_CHILD);
2866                 if (S_ISDIR(attr->la_mode)) {
2867                         /* Drop the reference, no need to delete "."/"..",
2868                          * because the object is to be destroyed directly. */
2869                         rc2 = mdo_ref_del(env, son, handle);
2870                         if (rc2 != 0) {
2871                                 mdd_write_unlock(env, son);
2872                                 goto out_stop;
2873                         }
2874                 }
2875 out_volatile:
2876                 /* For volatile files drop one link immediately, since there is
2877                  * no filename in the namespace, and save any error returned. */
2878                 rc2 = mdo_ref_del(env, son, handle);
2879                 if (rc2 != 0) {
2880                         mdd_write_unlock(env, son);
2881                         if (unlikely(rc == 0))
2882                                 rc = rc2;
2883                         goto out_stop;
2884                 }
2885
2886                 /* Don't destroy the volatile object on success */
2887                 if (likely(rc != 0))
2888                         mdo_destroy(env, son, handle);
2889                 mdd_write_unlock(env, son);
2890         }
2891
2892         if (rc == 0 && fid_is_namespace_visible(mdd_object_fid(son)) &&
2893             likely((spec->sp_cr_flags & MDS_OPEN_VOLATILE) == 0))
2894                 rc = mdd_changelog_ns_store(env, mdd,
2895                                 S_ISDIR(attr->la_mode) ? CL_MKDIR :
2896                                 S_ISREG(attr->la_mode) ? CL_CREATE :
2897                                 S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2898                                 0, son, mdd_object_fid(mdd_pobj), NULL, NULL,
2899                                 lname, NULL, handle);
2900 out_stop:
2901         rc2 = mdd_trans_stop(env, mdd, rc, handle);
2902         if (rc == 0) {
2903                 /* If creation fails, it is most likely due to the remote update
2904                  * failure, because local transaction will mostly succeed at
2905                  * this stage. There is no easy way to rollback all of previous
2906                  * updates, so let's remove the object from namespace, and
2907                  * LFSCK should handle the orphan object. */
2908                 if (rc2 < 0 && !mdd_object_remote(mdd_pobj))
2909                         mdd_index_delete(env, mdd_pobj, attr, lname);
2910                 rc = rc2;
2911         }
2912 out_free:
2913         if (is_vmalloc_addr(ldata->ld_buf))
2914                 /* if we vmalloced a large buffer drop it */
2915                 lu_buf_free(ldata->ld_buf);
2916
2917         if (spec->sp_cr_flags & MDS_OPEN_PCC)
2918                 lu_buf_free(&hsm_buf);
2919
2920         /* The child object shouldn't be cached anymore */
2921         if (rc)
2922                 set_bit(LU_OBJECT_HEARD_BANSHEE,
2923                         &child->mo_lu.lo_header->loh_flags);
2924         return rc;
2925 }
2926
2927 /* has not mdd_write{read}_lock on any obj yet. */
2928 static int mdd_rename_sanity_check(const struct lu_env *env,
2929                                    struct mdd_object *src_pobj,
2930                                    const struct lu_attr *pattr,
2931                                    struct mdd_object *tgt_pobj,
2932                                    const struct lu_attr *tpattr,
2933                                    struct mdd_object *sobj,
2934                                    const struct lu_attr *cattr,
2935                                    struct mdd_object *tobj,
2936                                    const struct lu_attr *tattr)
2937 {
2938         int rc = 0;
2939         ENTRY;
2940
2941         /* XXX: when get here, sobj must NOT be NULL,
2942          * the other case has been processed in cld_rename
2943          * before mdd_rename and enable MDS_PERM_BYPASS. */
2944         LASSERT(sobj);
2945
2946         /*
2947          * If we are using project inheritance, we only allow renames
2948          * into our tree when the project IDs are the same; otherwise
2949          * tree quota mechanism would be circumvented.
2950          */
2951         if ((((tpattr->la_flags & LUSTRE_PROJINHERIT_FL) &&
2952             tpattr->la_projid != cattr->la_projid) ||
2953             ((pattr->la_flags & LUSTRE_PROJINHERIT_FL) &&
2954             (pattr->la_projid != tpattr->la_projid))) &&
2955             S_ISDIR(cattr->la_mode))
2956                 RETURN(-EXDEV);
2957
2958         /* we prevent an encrypted file from being renamed
2959          * into an unencrypted dir
2960          */
2961         if ((pattr->la_valid & LA_FLAGS &&
2962              pattr->la_flags & LUSTRE_ENCRYPT_FL) &&
2963             !(tpattr->la_valid & LA_FLAGS &&
2964               tpattr->la_flags & LUSTRE_ENCRYPT_FL))
2965                 RETURN(-EXDEV);
2966
2967         rc = mdd_may_delete(env, src_pobj, pattr, sobj, cattr, NULL, 1, 0);
2968         if (rc)
2969                 RETURN(rc);
2970
2971         /* XXX: when get here, "tobj == NULL" means tobj must
2972          * NOT exist (neither on remote MDS, such case has been
2973          * processed in cld_rename before mdd_rename and enable
2974          * MDS_PERM_BYPASS).
2975          * So check may_create, but not check may_unlink. */
2976         if (tobj == NULL)
2977                 rc = mdd_may_create(env, tgt_pobj, tpattr, NULL,
2978                                     (src_pobj != tgt_pobj));
2979         else
2980                 rc = mdd_may_delete(env, tgt_pobj, tpattr, tobj, tattr, cattr,
2981                                     (src_pobj != tgt_pobj), 1);
2982
2983         if (!rc && !tobj && (src_pobj != tgt_pobj) && S_ISDIR(cattr->la_mode))
2984                 rc = __mdd_may_link(env, tgt_pobj, tpattr);
2985
2986         RETURN(rc);
2987 }
2988
2989 static int mdd_declare_rename(const struct lu_env *env,
2990                               struct mdd_device *mdd,
2991                               struct mdd_object *mdd_spobj,
2992                               struct mdd_object *mdd_tpobj,
2993                               struct mdd_object *mdd_sobj,
2994                               struct mdd_object *mdd_tobj,
2995                               const struct lu_name *sname,
2996                               const struct lu_name *tname,
2997                               struct md_attr *ma,
2998                               struct linkea_data *ldata, bool change_projid,
2999                               struct thandle *handle)
3000 {
3001         struct lu_attr *la = &mdd_env_info(env)->mdi_la_for_fix;
3002         int rc;
3003
3004         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
3005         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
3006
3007         LASSERT(mdd_spobj);
3008         LASSERT(mdd_tpobj);
3009         LASSERT(mdd_sobj);
3010
3011         /* name from source dir */
3012         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
3013         if (rc)
3014                 return rc;
3015
3016         /* .. from source child */
3017         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
3018                 /* source child can be directory, count by source dir's nlink */
3019                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
3020                 if (rc)
3021                         return rc;
3022                 if (mdd_spobj != mdd_tpobj) {
3023                         rc = mdo_declare_index_delete(env, mdd_sobj, dotdot,
3024                                                       handle);
3025                         if (rc != 0)
3026                                 return rc;
3027
3028                         rc = mdo_declare_index_insert(env, mdd_sobj,
3029                                                       mdd_object_fid(mdd_tpobj),
3030                                                       S_IFDIR, dotdot, handle);
3031                         if (rc != 0)
3032                                 return rc;
3033                 }
3034
3035                 /* new target child can be directory,
3036                  * counted by target dir's nlink */
3037                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
3038                 if (rc != 0)
3039                         return rc;
3040         }
3041
3042         la->la_valid = LA_CTIME | LA_MTIME;
3043         rc = mdo_declare_attr_set(env, mdd_spobj, la, handle);
3044         if (rc != 0)
3045                 return rc;
3046
3047         rc = mdo_declare_attr_set(env, mdd_tpobj, la, handle);
3048         if (rc != 0)
3049                 return rc;
3050
3051         la->la_valid = LA_CTIME;
3052         if (change_projid)
3053                 la->la_valid |= LA_PROJID;
3054         rc = mdo_declare_attr_set(env, mdd_sobj, la, handle);
3055         if (rc)
3056                 return rc;
3057
3058         rc = mdd_declare_links_add(env, mdd_sobj, handle, ldata);
3059         if (rc)
3060                 return rc;
3061
3062         /* new name */
3063         rc = mdo_declare_index_insert(env, mdd_tpobj, mdd_object_fid(mdd_sobj),
3064                                       mdd_object_type(mdd_sobj),
3065                                       tname->ln_name, handle);
3066         if (rc != 0)
3067                 return rc;
3068
3069         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
3070                 /* delete target child in target parent directory */
3071                 rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name,
3072                                               handle);
3073                 if (rc)
3074                         return rc;
3075
3076                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
3077                 if (rc)
3078                         return rc;
3079
3080                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
3081                         /* target child can be directory,
3082                          * delete "." reference in target child directory */
3083                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
3084                         if (rc)
3085                                 return rc;
3086
3087                         /* delete ".." reference in target parent directory */
3088                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
3089                         if (rc)
3090                                 return rc;
3091                 }
3092
3093                 la->la_valid = LA_CTIME;
3094                 rc = mdo_declare_attr_set(env, mdd_tobj, la, handle);
3095                 if (rc)
3096                         return rc;
3097
3098                 rc = mdd_declare_finish_unlink(env, mdd_tobj, handle);
3099                 if (rc)
3100                         return rc;
3101         }
3102
3103         rc = mdd_declare_changelog_store(env, mdd, CL_RENAME, tname, sname,
3104                                          handle);
3105         if (rc)
3106                 return rc;
3107
3108         return rc;
3109 }
3110
3111 static int mdd_migrate_object(const struct lu_env *env,
3112                               struct mdd_object *spobj,
3113                               struct mdd_object *tpobj,
3114                               struct mdd_object *sobj,
3115                               struct mdd_object *tobj,
3116                               const struct lu_name *sname,
3117                               const struct lu_name *tname,
3118                               struct md_op_spec *spec,
3119                               struct md_attr *ma);
3120
3121 /* src object can be remote that is why we use only fid and type of object */
3122 static int mdd_rename(const struct lu_env *env,
3123                       struct md_object *src_pobj, struct md_object *tgt_pobj,
3124                       const struct lu_fid *lf, const struct lu_name *lsname,
3125                       struct md_object *tobj, const struct lu_name *ltname,
3126                       struct md_attr *ma)
3127 {
3128         const char *sname = lsname->ln_name;
3129         const char *tname = ltname->ln_name;
3130         struct lu_attr    *la = &mdd_env_info(env)->mdi_la_for_fix;
3131         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
3132         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
3133         struct mdd_device *mdd = mdo2mdd(src_pobj);
3134         struct mdd_object *mdd_sobj = NULL;                  /* source object */
3135         struct mdd_object *mdd_tobj = NULL;
3136         struct lu_attr *cattr = MDD_ENV_VAR(env, cattr);
3137         struct lu_attr *pattr = MDD_ENV_VAR(env, pattr);
3138         struct lu_attr *tattr = MDD_ENV_VAR(env, tattr);
3139         struct lu_attr *tpattr = MDD_ENV_VAR(env, tpattr);
3140         struct thandle *handle;
3141         struct linkea_data  *ldata = &mdd_env_info(env)->mdi_link_data;
3142         const struct lu_fid *tpobj_fid = mdd_object_fid(mdd_tpobj);
3143         const struct lu_fid *spobj_fid = mdd_object_fid(mdd_spobj);
3144         bool is_dir;
3145         bool tobj_ref = 0;
3146         bool tobj_locked = 0;
3147         bool change_projid = false;
3148         unsigned cl_flags = 0;
3149         int rc, rc2;
3150         ENTRY;
3151
3152         /* let unlink to complete and commit */
3153         CFS_FAIL_TIMEOUT(OBD_FAIL_TGT_REPLY_DATA_RACE, 2 + cfs_fail_val);
3154
3155         if (tobj)
3156                 mdd_tobj = md2mdd_obj(tobj);
3157
3158         mdd_sobj = mdd_object_find(env, mdd, lf);
3159         if (IS_ERR(mdd_sobj))
3160                 RETURN(PTR_ERR(mdd_sobj));
3161
3162         rc = mdd_la_get(env, mdd_sobj, cattr);
3163         if (rc)
3164                 GOTO(out_pending, rc);
3165
3166         /* if rename is cross MDTs, migrate symlink if it doesn't have other
3167          * hard links, and target doesn't exist.
3168          */
3169         if (mdd_object_remote(mdd_sobj) && S_ISLNK(cattr->la_mode) &&
3170             cattr->la_nlink == 1 && !tobj) {
3171                 struct md_op_spec *spec = &mdd_env_info(env)->mdi_spec;
3172                 struct lu_device *ld = &mdd->mdd_md_dev.md_lu_dev;
3173                 struct lu_fid tfid;
3174
3175                 rc = ld->ld_ops->ldo_fid_alloc(env, ld, &tfid, &tgt_pobj->mo_lu,
3176                                                NULL);
3177                 if (rc < 0)
3178                         GOTO(out_pending, rc);
3179
3180                 mdd_tobj = mdd_object_find(env, mdd, &tfid);
3181                 if (IS_ERR(mdd_tobj))
3182                         GOTO(out_pending, rc = PTR_ERR(mdd_tobj));
3183
3184                 memset(spec, 0, sizeof(*spec));
3185                 rc = mdd_migrate_object(env, mdd_spobj, mdd_tpobj, mdd_sobj,
3186                                         mdd_tobj, lsname, ltname, spec, ma);
3187                 mdd_object_put(env, mdd_tobj);
3188                 GOTO(out_pending, rc);
3189         }
3190
3191         rc = mdd_la_get(env, mdd_spobj, pattr);
3192         if (rc)
3193                 GOTO(out_pending, rc);
3194
3195         if (mdd_tobj) {
3196                 rc = mdd_la_get(env, mdd_tobj, tattr);
3197                 if (rc)
3198                         GOTO(out_pending, rc);
3199                 /* search for an existing archive.
3200                  * we should check ahead as the object
3201                  * can be destroyed in this transaction */
3202                 if (mdd_hsm_archive_exists(env, mdd_tobj, ma))
3203                         cl_flags |= CLF_RENAME_LAST_EXISTS;
3204         }
3205
3206         rc = mdd_la_get(env, mdd_tpobj, tpattr);
3207         if (rc)
3208                 GOTO(out_pending, rc);
3209
3210         rc = mdd_rename_sanity_check(env, mdd_spobj, pattr, mdd_tpobj, tpattr,
3211                                      mdd_sobj, cattr, mdd_tobj, tattr);
3212         if (rc)
3213                 GOTO(out_pending, rc);
3214
3215         rc = mdd_name_check(env, mdd, ltname);
3216         if (rc < 0)
3217                 GOTO(out_pending, rc);
3218
3219         handle = mdd_trans_create(env, mdd);
3220         if (IS_ERR(handle))
3221                 GOTO(out_pending, rc = PTR_ERR(handle));
3222
3223         memset(ldata, 0, sizeof(*ldata));
3224         rc = mdd_linkea_prepare(env, mdd_sobj, mdd_object_fid(mdd_spobj),
3225                                 lsname, mdd_object_fid(mdd_tpobj), ltname,
3226                                 1, 0, ldata);
3227         if (rc)
3228                 GOTO(stop, rc);
3229
3230         if (tpattr->la_projid != cattr->la_projid &&
3231             tpattr->la_flags & LUSTRE_PROJINHERIT_FL)
3232                 change_projid = true;
3233
3234         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
3235                                 mdd_tobj, lsname, ltname, ma, ldata,
3236                                 change_projid, handle);
3237         if (rc)
3238                 GOTO(stop, rc);
3239
3240         rc = mdd_trans_start(env, mdd, handle);
3241         if (rc)
3242                 GOTO(stop, rc);
3243
3244         is_dir = S_ISDIR(cattr->la_mode);
3245
3246         /* Remove source name from source directory */
3247         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle);
3248         if (rc != 0)
3249                 GOTO(stop, rc);
3250
3251         /* "mv dir1 dir2" needs "dir1/.." link update */
3252         if (is_dir && !lu_fid_eq(spobj_fid, tpobj_fid)) {
3253                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3254                 if (rc != 0)
3255                         GOTO(fixup_spobj2, rc);
3256
3257                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, S_IFDIR,
3258                                              dotdot, handle);
3259                 if (rc != 0)
3260                         GOTO(fixup_spobj, rc);
3261         }
3262
3263         if (mdd_tobj != NULL && mdd_object_exists(mdd_tobj)) {
3264                 rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
3265                 if (rc != 0)
3266                         /* tname might been renamed to something else */
3267                         GOTO(fixup_spobj, rc);
3268         }
3269
3270         /* Insert new fid with target name into target dir */
3271         rc = __mdd_index_insert(env, mdd_tpobj, lf, cattr->la_mode,
3272                                 tname, handle);
3273         if (rc != 0)
3274                 GOTO(fixup_tpobj, rc);
3275
3276         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
3277         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
3278
3279         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
3280         la->la_valid = LA_CTIME;
3281         if (change_projid) {
3282                 /* mdd_update_time honors other valid flags except TIME ones */
3283                 la->la_valid |= LA_PROJID;
3284                 la->la_projid = tpattr->la_projid;
3285         }
3286         rc = mdd_update_time(env, mdd_sobj, cattr, la, handle);
3287         if (rc)
3288                 GOTO(fixup_tpobj, rc);
3289
3290         /* Update the linkEA for the source object */
3291         mdd_write_lock(env, mdd_sobj, DT_SRC_CHILD);
3292         rc = mdd_links_rename(env, mdd_sobj, mdd_object_fid(mdd_spobj),
3293                               lsname, mdd_object_fid(mdd_tpobj), ltname,
3294                               handle, ldata, 0, 0);
3295         if (rc == -ENOENT)
3296                 /* Old files might not have EA entry */
3297                 mdd_links_add(env, mdd_sobj, mdd_object_fid(mdd_spobj),
3298                               lsname, handle, NULL, 0);
3299         mdd_write_unlock(env, mdd_sobj);
3300         /* We don't fail the transaction if the link ea can't be
3301            updated -- fid2path will use alternate lookup method. */
3302         rc = 0;
3303
3304         /* Remove old target object
3305          * For tobj is remote case cmm layer has processed
3306          * and set tobj to NULL then. So when tobj is NOT NULL,
3307          * it must be local one.
3308          */
3309         if (tobj && mdd_object_exists(mdd_tobj)) {
3310                 mdd_write_lock(env, mdd_tobj, DT_TGT_CHILD);
3311                 tobj_locked = 1;
3312                 if (mdd_is_dead_obj(mdd_tobj)) {
3313                         /* shld not be dead, something is wrong */
3314                         CERROR("tobj is dead, something is wrong\n");
3315                         rc = -EINVAL;
3316                         goto cleanup;
3317                 }
3318                 mdo_ref_del(env, mdd_tobj, handle);
3319
3320                 /* Remove dot reference. */
3321                 if (S_ISDIR(tattr->la_mode))
3322                         mdo_ref_del(env, mdd_tobj, handle);
3323                 tobj_ref = 1;
3324
3325                 /* fetch updated nlink */
3326                 rc = mdd_la_get(env, mdd_tobj, tattr);
3327                 if (rc != 0) {
3328                         CERROR("%s: Failed to get nlink for tobj "
3329                                 DFID": rc = %d\n",
3330                                 mdd2obd_dev(mdd)->obd_name,
3331                                 PFID(tpobj_fid), rc);
3332                         GOTO(fixup_tpobj, rc);
3333                 }
3334
3335                 la->la_valid = LA_CTIME;
3336                 rc = mdd_update_time(env, mdd_tobj, tattr, la, handle);
3337                 if (rc != 0) {
3338                         CERROR("%s: Failed to set ctime for tobj "
3339                                 DFID": rc = %d\n",
3340                                 mdd2obd_dev(mdd)->obd_name,
3341                                 PFID(tpobj_fid), rc);
3342                         GOTO(fixup_tpobj, rc);
3343                 }
3344
3345                 /* XXX: this transfer to ma will be removed with LOD/OSP */
3346                 ma->ma_attr = *tattr;
3347                 ma->ma_valid |= MA_INODE;
3348                 rc = mdd_finish_unlink(env, mdd_tobj, ma, mdd_tpobj, ltname,
3349                                        handle);
3350                 if (rc != 0) {
3351                         CERROR("%s: Failed to unlink tobj "
3352                                 DFID": rc = %d\n",
3353                                 mdd2obd_dev(mdd)->obd_name,
3354                                 PFID(tpobj_fid), rc);
3355                         GOTO(fixup_tpobj, rc);
3356                 }
3357
3358                 /* fetch updated nlink */
3359                 rc = mdd_la_get(env, mdd_tobj, tattr);
3360                 if (rc == -ENOENT) {
3361                         /* the object got removed, let's
3362                          * return the latest known attributes */
3363                         tattr->la_nlink = 0;
3364                         rc = 0;
3365                 } else if (rc != 0) {
3366                         CERROR("%s: Failed to get nlink for tobj "
3367                                 DFID": rc = %d\n",
3368                                 mdd2obd_dev(mdd)->obd_name,
3369                                 PFID(tpobj_fid), rc);
3370                         GOTO(fixup_tpobj, rc);
3371                 }
3372                 /* XXX: this transfer to ma will be removed with LOD/OSP */
3373                 ma->ma_attr = *tattr;
3374                 ma->ma_valid |= MA_INODE;
3375
3376                 if (tattr->la_nlink == 0)
3377                         cl_flags |= CLF_RENAME_LAST;
3378                 else
3379                         cl_flags &= ~CLF_RENAME_LAST_EXISTS;
3380         }
3381
3382         la->la_valid = LA_CTIME | LA_MTIME;
3383         rc = mdd_update_time(env, mdd_spobj, pattr, la, handle);
3384         if (rc)
3385                 GOTO(fixup_tpobj, rc);
3386
3387         if (mdd_spobj != mdd_tpobj) {
3388                 la->la_valid = LA_CTIME | LA_MTIME;
3389                 rc = mdd_update_time(env, mdd_tpobj, tpattr, la, handle);
3390                 if (rc != 0)
3391                         GOTO(fixup_tpobj, rc);
3392         }
3393
3394         EXIT;
3395
3396 fixup_tpobj:
3397         if (rc) {
3398                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle);
3399                 if (rc2)
3400                         CWARN("tp obj fix error %d\n",rc2);
3401
3402                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
3403                     !mdd_is_dead_obj(mdd_tobj)) {
3404                         if (tobj_ref) {
3405                                 mdo_ref_add(env, mdd_tobj, handle);
3406                                 if (is_dir)
3407                                         mdo_ref_add(env, mdd_tobj, handle);
3408                         }
3409
3410                         rc2 = __mdd_index_insert(env, mdd_tpobj,
3411                                                  mdd_object_fid(mdd_tobj),
3412                                                  mdd_object_type(mdd_tobj),
3413                                                  tname, handle);
3414                         if (rc2 != 0)
3415                                 CWARN("tp obj fix error: rc = %d\n", rc2);
3416                 }
3417         }
3418
3419 fixup_spobj:
3420         if (rc && is_dir && mdd_sobj && mdd_spobj != mdd_tpobj) {
3421                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3422                 if (rc2)
3423                         CWARN("%s: sp obj dotdot delete error: rc = %d\n",
3424                                mdd2obd_dev(mdd)->obd_name, rc2);
3425
3426
3427                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid, S_IFDIR,
3428                                               dotdot, handle);
3429                 if (rc2 != 0)
3430                         CWARN("%s: sp obj dotdot insert error: rc = %d\n",
3431                               mdd2obd_dev(mdd)->obd_name, rc2);
3432         }
3433
3434 fixup_spobj2:
3435         if (rc != 0) {
3436                 rc2 = __mdd_index_insert(env, mdd_spobj, lf,
3437                                          mdd_object_type(mdd_sobj), sname,
3438                                          handle);
3439                 if (rc2 != 0)
3440                         CWARN("sp obj fix error: rc = %d\n", rc2);
3441         }
3442
3443 cleanup:
3444         if (tobj_locked)
3445                 mdd_write_unlock(env, mdd_tobj);
3446
3447         if (rc == 0)
3448                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, cl_flags,
3449                                             mdd_tobj, tpobj_fid, lf, spobj_fid,
3450                                             ltname, lsname, handle);
3451
3452 stop:
3453         rc = mdd_trans_stop(env, mdd, rc, handle);
3454
3455 out_pending:
3456         mdd_object_put(env, mdd_sobj);
3457         return rc;
3458 }
3459
3460 /**
3461  * Check whether we should migrate the file/dir
3462  * return val
3463  *      < 0  permission check failed or other error.
3464  *      = 0  the file can be migrated.
3465  **/
3466 static int mdd_migrate_sanity_check(const struct lu_env *env,
3467                                     struct mdd_device *mdd,
3468                                     struct mdd_object *spobj,
3469                                     struct mdd_object *tpobj,
3470                                     struct mdd_object *sobj,
3471                                     struct mdd_object *tobj,
3472                                     const struct lu_attr *spattr,
3473                                     const struct lu_attr *tpattr,
3474                                     const struct lu_attr *attr)
3475 {
3476         int rc;
3477
3478         ENTRY;
3479
3480         if (!mdd_object_remote(sobj)) {
3481                 mdd_read_lock(env, sobj, DT_SRC_CHILD);
3482                 if (sobj->mod_count > 0) {
3483                         CDEBUG(D_INFO, "%s: "DFID" is opened, count %d\n",
3484                                mdd_obj_dev_name(sobj),
3485                                PFID(mdd_object_fid(sobj)),
3486                                sobj->mod_count);
3487                         mdd_read_unlock(env, sobj);
3488                         RETURN(-EBUSY);
3489                 }
3490                 mdd_read_unlock(env, sobj);
3491         }
3492
3493         if (mdd_object_exists(tobj))
3494                 RETURN(-EEXIST);
3495
3496         rc = mdd_may_delete(env, spobj, spattr, sobj, attr, NULL, 1, 0);
3497         if (rc)
3498                 RETURN(rc);
3499
3500         rc = mdd_may_create(env, tpobj, tpattr, NULL, true);
3501
3502         RETURN(rc);
3503 }
3504
3505 struct mdd_xattr_entry {
3506         struct list_head        mxe_linkage;
3507         char                   *mxe_name;
3508         struct lu_buf           mxe_buf;
3509 };
3510
3511 struct mdd_xattrs {
3512         struct lu_buf           mx_namebuf;
3513         struct list_head        mx_list;
3514 };
3515
3516 static inline void mdd_xattrs_init(struct mdd_xattrs *xattrs)
3517 {
3518         INIT_LIST_HEAD(&xattrs->mx_list);
3519         xattrs->mx_namebuf.lb_buf = NULL;
3520         xattrs->mx_namebuf.lb_len = 0;
3521 }
3522
3523 static inline void mdd_xattrs_fini(struct mdd_xattrs *xattrs)
3524 {
3525         struct mdd_xattr_entry *entry;
3526         struct mdd_xattr_entry *tmp;
3527
3528         list_for_each_entry_safe(entry, tmp, &xattrs->mx_list, mxe_linkage) {
3529                 lu_buf_free(&entry->mxe_buf);
3530                 list_del(&entry->mxe_linkage);
3531                 OBD_FREE_PTR(entry);
3532         }
3533
3534         lu_buf_free(&xattrs->mx_namebuf);
3535 }
3536
3537 /* read xattrs into buf, but skip LMA, LMV, LINKEA if 'skip_linkea' is
3538  * set, and DMV if 'skip_dmv" is set.
3539  */
3540 static int mdd_xattrs_migrate_prep(const struct lu_env *env,
3541                                    struct mdd_xattrs *xattrs,
3542                                    struct mdd_object *sobj,
3543                                    bool skip_linkea,
3544                                    bool skip_dmv)
3545 {
3546         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
3547         struct mdd_xattr_entry *entry;
3548         bool needencxattr = false;
3549         bool encxattrfound = false;
3550         char *xname;
3551         int list_xsize;
3552         int xlen;
3553         int rem;
3554         int xsize;
3555         int rc;
3556
3557         ENTRY;
3558
3559         list_xsize = mdo_xattr_list(env, sobj, &LU_BUF_NULL);
3560         if (list_xsize == -ENODATA)
3561                 RETURN(0);
3562
3563         if (list_xsize < 0)
3564                 RETURN(list_xsize);
3565
3566         if (attr->la_valid & LA_FLAGS &&
3567             attr->la_flags & LUSTRE_ENCRYPT_FL) {
3568                 needencxattr = true;
3569                 list_xsize +=
3570                         strlen(LL_XATTR_NAME_ENCRYPTION_CONTEXT) + 1;
3571         }
3572
3573         lu_buf_alloc(&xattrs->mx_namebuf, list_xsize);
3574         if (xattrs->mx_namebuf.lb_buf == NULL)
3575                 RETURN(-ENOMEM);
3576
3577         rc = mdo_xattr_list(env, sobj, &xattrs->mx_namebuf);
3578         if (rc < 0)
3579                 GOTO(fini, rc);
3580
3581         rem = rc;
3582         rc = 0;
3583         xname = xattrs->mx_namebuf.lb_buf;
3584 reloop:
3585         for (; rem > 0; xname += xlen, rem -= xlen) {
3586                 if (needencxattr &&
3587                     strcmp(xname, LL_XATTR_NAME_ENCRYPTION_CONTEXT) == 0)
3588                         encxattrfound = true;
3589                 xlen = strnlen(xname, rem - 1) + 1;
3590                 if (strcmp(XATTR_NAME_LMA, xname) == 0 ||
3591                     strcmp(XATTR_NAME_LMV, xname) == 0)
3592                         continue;
3593
3594                 if (skip_linkea &&
3595                     strcmp(XATTR_NAME_LINK, xname) == 0)
3596                         continue;
3597
3598                 if (skip_dmv &&
3599                     strcmp(XATTR_NAME_DEFAULT_LMV, xname) == 0)
3600                         continue;
3601
3602                 xsize = mdo_xattr_get(env, sobj, &LU_BUF_NULL, xname);
3603                 if (xsize == -ENODATA)
3604                         continue;
3605                 if (xsize < 0)
3606                         GOTO(fini, rc = xsize);
3607
3608                 OBD_ALLOC_PTR(entry);
3609                 if (!entry)
3610                         GOTO(fini, rc = -ENOMEM);
3611
3612                 lu_buf_alloc(&entry->mxe_buf, xsize);
3613                 if (!entry->mxe_buf.lb_buf) {
3614                         OBD_FREE_PTR(entry);
3615                         GOTO(fini, rc = -ENOMEM);
3616                 }
3617
3618                 rc = mdo_xattr_get(env, sobj, &entry->mxe_buf, xname);
3619                 if (rc < 0) {
3620                         lu_buf_free(&entry->mxe_buf);
3621                         OBD_FREE_PTR(entry);
3622                         if (rc == -ENODATA)
3623                                 continue;
3624                         GOTO(fini, rc);
3625                 }
3626
3627                 entry->mxe_name = xname;
3628                 list_add_tail(&entry->mxe_linkage, &xattrs->mx_list);
3629         }
3630
3631         if (needencxattr && !encxattrfound) {
3632                 xlen = strlen(LL_XATTR_NAME_ENCRYPTION_CONTEXT) + 1;
3633                 strncpy(xname, LL_XATTR_NAME_ENCRYPTION_CONTEXT, xlen);
3634                 rem = xlen;
3635                 GOTO(reloop, 0);
3636         }
3637
3638         RETURN(0);
3639 fini:
3640         mdd_xattrs_fini(xattrs);
3641         RETURN(rc);
3642 }
3643
3644 typedef int (*mdd_xattr_cb)(const struct lu_env *env,
3645                             struct mdd_object *obj,
3646                             const struct lu_buf *buf,
3647                             const char *name,
3648                             int fl, struct thandle *handle);
3649
3650 static int mdd_foreach_xattr(const struct lu_env *env,
3651                              struct mdd_object *tobj,
3652                              struct mdd_xattrs *xattrs,
3653                              struct thandle *handle,
3654                              mdd_xattr_cb cb)
3655 {
3656         struct mdd_xattr_entry *entry;
3657         int rc;
3658
3659         list_for_each_entry(entry, &xattrs->mx_list, mxe_linkage) {
3660                 rc = cb(env, tobj, &entry->mxe_buf, entry->mxe_name, 0, handle);
3661                 if (rc)
3662                         return rc;
3663         }
3664
3665         return 0;
3666 }
3667
3668 typedef int (*mdd_linkea_cb)(const struct lu_env *env,
3669                              struct mdd_object *sobj,
3670                              struct mdd_object *tobj,
3671                              const struct lu_name *sname,
3672                              const struct lu_fid *sfid,
3673                              const struct lu_name *lname,
3674                              const struct lu_fid *fid,
3675                              void *opaque,
3676                              struct thandle *handle);
3677
3678 static int mdd_declare_update_link(const struct lu_env *env,
3679                                    struct mdd_object *sobj,
3680                                    struct mdd_object *tobj,
3681                                    const struct lu_name *tname,
3682                                    const struct lu_fid *tpfid,
3683                                    const struct lu_name *lname,
3684                                    const struct lu_fid *fid,
3685                                    void *unused,
3686                                    struct thandle *handle)
3687 {
3688         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3689         struct mdd_object *pobj;
3690         int rc;
3691
3692         /* ignore tobj */
3693         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3694             !strcmp(tname->ln_name, lname->ln_name))
3695                 return 0;
3696
3697         pobj = mdd_object_find(env, mdd, fid);
3698         if (IS_ERR(pobj))
3699                 return PTR_ERR(pobj);
3700
3701
3702         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
3703         if (!rc)
3704                 rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(tobj),
3705                                               mdd_object_type(sobj),
3706                                               lname->ln_name, handle);
3707         mdd_object_put(env, pobj);
3708         if (rc)
3709                 return rc;
3710
3711         rc = mdo_declare_ref_add(env, tobj, handle);
3712         if (rc)
3713                 return rc;
3714
3715         rc = mdo_declare_ref_del(env, sobj, handle);
3716         return rc;
3717 }
3718
3719 static int mdd_update_link(const struct lu_env *env,
3720                            struct mdd_object *sobj,
3721                            struct mdd_object *tobj,
3722                            const struct lu_name *tname,
3723                            const struct lu_fid *tpfid,
3724                            const struct lu_name *lname,
3725                            const struct lu_fid *fid,
3726                            void *unused,
3727                            struct thandle *handle)
3728 {
3729         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3730         struct mdd_object *pobj;
3731         int rc;
3732
3733         ENTRY;
3734
3735         /* ignore tobj */
3736         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3737             !memcmp(tname->ln_name, lname->ln_name, lname->ln_namelen))
3738                 RETURN(0);
3739
3740         CDEBUG(D_INFO, "update "DFID"/"DNAME":"DFID"\n",
3741                PFID(fid), PNAME(lname), PFID(mdd_object_fid(tobj)));
3742
3743         pobj = mdd_object_find(env, mdd, fid);
3744         if (IS_ERR(pobj)) {
3745                 CWARN("%s: cannot find obj "DFID": %ld\n",
3746                       mdd2obd_dev(mdd)->obd_name, PFID(fid), PTR_ERR(pobj));
3747                 RETURN(PTR_ERR(pobj));
3748         }
3749
3750         if (!mdd_object_exists(pobj)) {
3751                 CDEBUG(D_INFO, DFID" doesn't exist\n", PFID(fid));
3752                 mdd_object_put(env, pobj);
3753                 RETURN(-ENOENT);
3754         }
3755
3756         mdd_write_lock(env, pobj, DT_TGT_PARENT);
3757         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
3758         if (!rc)
3759                 rc = __mdd_index_insert_only(env, pobj, mdd_object_fid(tobj),
3760                                              mdd_object_type(sobj),
3761                                              lname->ln_name, handle);
3762         mdd_write_unlock(env, pobj);
3763         mdd_object_put(env, pobj);
3764         if (rc)
3765                 RETURN(rc);
3766
3767         mdd_write_lock(env, tobj, DT_TGT_CHILD);
3768         rc = mdo_ref_add(env, tobj, handle);
3769         mdd_write_unlock(env, tobj);
3770         if (rc)
3771                 RETURN(rc);
3772
3773         mdd_write_lock(env, sobj, DT_SRC_CHILD);
3774         rc = mdo_ref_del(env, sobj, handle);
3775         mdd_write_unlock(env, sobj);
3776
3777         RETURN(rc);
3778 }
3779
3780 static inline int mdd_fld_lookup(const struct lu_env *env,
3781                                  struct mdd_device *mdd,
3782                                  const struct lu_fid *fid,
3783                                  __u32 *mdt_index)
3784 {
3785         struct lu_seq_range *range = &mdd_env_info(env)->mdi_range;
3786         struct seq_server_site *ss;
3787         int rc;
3788
3789         ss = mdd->mdd_md_dev.md_lu_dev.ld_site->ld_seq_site;
3790
3791         range->lsr_flags = LU_SEQ_RANGE_MDT;
3792         rc = fld_server_lookup(env, ss->ss_server_fld, fid->f_seq, range);
3793         if (rc)
3794                 return rc;
3795
3796         *mdt_index = range->lsr_index;
3797
3798         return 0;
3799 }
3800
3801 static int mdd_is_link_on_source_mdt(const struct lu_env *env,
3802                                      struct mdd_object *sobj,
3803                                      struct mdd_object *tobj,
3804                                      const struct lu_name *tname,
3805                                      const struct lu_fid *tpfid,
3806                                      const struct lu_name *lname,
3807                                      const struct lu_fid *fid,
3808                                      void *opaque,
3809                                      struct thandle *handle)
3810 {
3811         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3812         __u32 source_mdt_index = *(__u32 *)opaque;
3813         __u32 link_mdt_index;
3814         int rc;
3815
3816         ENTRY;
3817
3818         /* ignore tobj */
3819         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3820             !memcmp(tname->ln_name, lname->ln_name, lname->ln_namelen))
3821                 return 0;
3822
3823         rc = mdd_fld_lookup(env, mdd, fid, &link_mdt_index);
3824         if (rc)
3825                 RETURN(rc);
3826
3827         RETURN(link_mdt_index == source_mdt_index);
3828 }
3829
3830 static int mdd_iterate_linkea(const struct lu_env *env,
3831                               struct mdd_object *sobj,
3832                               struct mdd_object *tobj,
3833                               const struct lu_name *tname,
3834                               const struct lu_fid *tpfid,
3835                               struct linkea_data *ldata,
3836                               void *opaque,
3837                               struct thandle *handle,
3838                               mdd_linkea_cb cb)
3839 {
3840         struct mdd_thread_info *info = mdd_env_info(env);
3841         char *filename = info->mdi_name;
3842         struct lu_name lname;
3843         struct lu_fid fid;
3844         int rc = 0;
3845
3846         if (!ldata->ld_buf)
3847                 return 0;
3848
3849         for (linkea_first_entry(ldata); ldata->ld_lee && !rc;
3850              linkea_next_entry(ldata)) {
3851                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
3852                                     &fid);
3853
3854                 /* Note: lname might miss \0 at the end */
3855                 snprintf(filename, sizeof(info->mdi_name), "%.*s",
3856                          lname.ln_namelen, lname.ln_name);
3857                 lname.ln_name = filename;
3858
3859                 CDEBUG(D_INFO, DFID"/"DNAME"\n", PFID(&fid), PNAME(&lname));
3860
3861                 rc = cb(env, sobj, tobj, tname, tpfid, &lname, &fid, opaque,
3862                         handle);
3863         }
3864
3865         return rc;
3866 }
3867
3868 /**
3869  * Prepare linkea, and check whether file needs migrate: if source still has
3870  * link on source MDT, no need to migrate, just update namespace on source and
3871  * target parents.
3872  *
3873  * \retval      0 do migrate
3874  * \retval      1 don't migrate
3875  * \retval      -errno on failure
3876  */
3877 static int mdd_migrate_linkea_prepare(const struct lu_env *env,
3878                                       struct mdd_device *mdd,
3879                                       struct mdd_object *spobj,
3880                                       struct mdd_object *tpobj,
3881                                       struct mdd_object *sobj,
3882                                       const struct lu_name *sname,
3883                                       const struct lu_name *tname,
3884                                       const struct lu_attr *attr,
3885                                       struct linkea_data *ldata)
3886 {
3887         __u32 source_mdt_index;
3888         int rc;
3889
3890         ENTRY;
3891
3892         memset(ldata, 0, sizeof(*ldata));
3893         rc = mdd_linkea_prepare(env, sobj, mdd_object_fid(spobj), sname,
3894                                 mdd_object_fid(tpobj), tname, 1, 0, ldata);
3895         if (rc)
3896                 RETURN(rc);
3897
3898         /*
3899          * Then it will check if the file should be migrated. If the file has
3900          * mulitple links, we only need migrate the file if all of its entries
3901          * has been migrated to the remote MDT.
3902          */
3903         if (S_ISDIR(attr->la_mode) || attr->la_nlink < 2)
3904                 RETURN(0);
3905
3906         /* If there are still links locally, don't migrate this file */
3907         LASSERT(ldata->ld_leh != NULL);
3908
3909         /*
3910          * If linkEA is overflow, it means there are some unknown name entries
3911          * under unknown parents, which will prevent the migration.
3912          */
3913         if (unlikely(ldata->ld_leh->leh_overflow_time))
3914                 RETURN(-EOVERFLOW);
3915
3916         rc = mdd_fld_lookup(env, mdd, mdd_object_fid(sobj), &source_mdt_index);
3917         if (rc)
3918                 RETURN(rc);
3919
3920         rc = mdd_iterate_linkea(env, sobj, NULL, tname, mdd_object_fid(tpobj),
3921                                 ldata, &source_mdt_index, NULL,
3922                                 mdd_is_link_on_source_mdt);
3923         RETURN(rc);
3924 }
3925
3926 static int mdd_declare_migrate_update(const struct lu_env *env,
3927                                       struct mdd_object *spobj,
3928                                       struct mdd_object *tpobj,
3929                                       struct mdd_object *obj,
3930                                       const struct lu_name *sname,
3931                                       const struct lu_name *tname,
3932                                       struct lu_attr *attr,
3933                                       struct lu_attr *spattr,
3934                                       struct lu_attr *tpattr,
3935                                       struct linkea_data *ldata,
3936                                       struct md_attr *ma,
3937                                       struct thandle *handle)
3938 {
3939         struct mdd_thread_info *info = mdd_env_info(env);
3940         struct lu_attr *la = &info->mdi_la_for_fix;
3941         int rc;
3942
3943         rc = mdo_declare_index_delete(env, spobj, sname->ln_name, handle);
3944         if (rc)
3945                 return rc;
3946
3947         if (S_ISDIR(attr->la_mode)) {
3948                 rc = mdo_declare_ref_del(env, spobj, handle);
3949                 if (rc)
3950                         return rc;
3951         }
3952
3953         rc = mdo_declare_index_insert(env, tpobj, mdd_object_fid(obj),
3954                                       attr->la_mode & S_IFMT,
3955                                       tname->ln_name, handle);
3956         if (rc)
3957                 return rc;
3958
3959         rc = mdd_declare_links_add(env, obj, handle, ldata);
3960         if (rc)
3961                 return rc;
3962
3963         if (S_ISDIR(attr->la_mode)) {
3964                 rc = mdo_declare_ref_add(env, tpobj, handle);
3965                 if (rc)
3966                         return rc;
3967         }
3968
3969         la->la_valid = LA_CTIME | LA_MTIME;
3970         rc = mdo_declare_attr_set(env, spobj, la, handle);
3971         if (rc)
3972                 return rc;
3973
3974         if (tpobj != spobj) {
3975                 rc = mdo_declare_attr_set(env, tpobj, la, handle);
3976                 if (rc)
3977                         return rc;
3978         }
3979
3980         return rc;
3981 }
3982
3983 static int mdd_declare_migrate_create(const struct lu_env *env,
3984                                       struct mdd_object *spobj,
3985                                       struct mdd_object *tpobj,
3986                                       struct mdd_object *sobj,
3987                                       struct mdd_object *tobj,
3988                                       const struct lu_name *sname,
3989                                       const struct lu_name *tname,
3990                                       struct lu_attr *spattr,
3991                                       struct lu_attr *tpattr,
3992                                       struct lu_attr *attr,
3993                                       struct lu_buf *sbuf,
3994                                       struct linkea_data *ldata,
3995                                       struct mdd_xattrs *xattrs,
3996                                       struct md_attr *ma,
3997                                       struct md_op_spec *spec,
3998                                       struct dt_allocation_hint *hint,
3999                                       struct thandle *handle)
4000 {
4001         struct mdd_thread_info *info = mdd_env_info(env);
4002         struct md_layout_change *mlc = &info->mdi_mlc;
4003         struct lmv_mds_md_v1 *lmv = sbuf->lb_buf;
4004         int rc;
4005
4006         ENTRY;
4007
4008         if (S_ISDIR(attr->la_mode)) {
4009                 struct lmv_user_md *lum = spec->u.sp_ea.eadata;
4010
4011                 mlc->mlc_opc = MD_LAYOUT_DETACH;
4012                 rc = mdo_declare_layout_change(env, sobj, mlc, handle);
4013                 if (rc)
4014                         return rc;
4015
4016                 lum->lum_hash_type |= cpu_to_le32(LMV_HASH_FLAG_MIGRATION);
4017         } else if (S_ISLNK(attr->la_mode)) {
4018                 spec->u.sp_symname.ln_name = sbuf->lb_buf;
4019                 /* don't count NUL */
4020                 spec->u.sp_symname.ln_namelen = sbuf->lb_len - 1;
4021         } else if (S_ISREG(attr->la_mode)) {
4022                 spec->sp_cr_flags |= MDS_OPEN_DELAY_CREATE;
4023                 spec->sp_cr_flags &= ~MDS_OPEN_HAS_EA;
4024         }
4025
4026         mdd_object_make_hint(env, tpobj, tobj, attr, spec, hint);
4027
4028         rc = mdd_declare_create(env, mdo2mdd(&tpobj->mod_obj), tpobj, tobj,
4029                                 tname, attr, handle, spec, ldata, NULL, NULL,
4030                                 NULL, hint);
4031         if (rc)
4032                 return rc;
4033
4034         /*
4035          * tobj mode will be used in mdo_declare_layout_change(), but it's not
4036          * createb yet, copy from sobj.
4037          */
4038         tobj->mod_obj.mo_lu.lo_header->loh_attr &= ~S_IFMT;
4039         tobj->mod_obj.mo_lu.lo_header->loh_attr |=
4040                 sobj->mod_obj.mo_lu.lo_header->loh_attr & S_IFMT;
4041
4042         if (S_ISDIR(attr->la_mode)) {
4043                 if (!lmv) {
4044                         /* if sobj is not striped, fake a 1-stripe LMV */
4045                         LASSERT(sizeof(info->mdi_key) >
4046                                 lmv_mds_md_size(1, LMV_MAGIC_V1));
4047                         lmv = (typeof(lmv))info->mdi_key;
4048                         memset(lmv, 0, sizeof(*lmv));
4049                         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
4050                         lmv->lmv_stripe_count = cpu_to_le32(1);
4051                         lmv->lmv_hash_type = cpu_to_le32(LMV_HASH_TYPE_DEFAULT);
4052                         fid_le_to_cpu(&lmv->lmv_stripe_fids[0],
4053                                       mdd_object_fid(sobj));
4054                         mlc->mlc_buf.lb_buf = lmv;
4055                         mlc->mlc_buf.lb_len = lmv_mds_md_size(1, LMV_MAGIC_V1);
4056                 } else {
4057                         mlc->mlc_buf = *sbuf;
4058                 }
4059                 mlc->mlc_opc = MD_LAYOUT_ATTACH;
4060                 rc = mdo_declare_layout_change(env, tobj, mlc, handle);
4061                 if (rc)
4062                         return rc;
4063         }
4064
4065         rc = mdd_foreach_xattr(env, tobj, xattrs, handle,
4066                                mdo_declare_xattr_set);
4067         if (rc)
4068                 return rc;
4069
4070         if (S_ISREG(attr->la_mode)) {
4071                 struct lu_buf fid_buf;
4072
4073                 handle->th_complex = 1;
4074
4075                 /* target may be remote, update PFID via sobj. */
4076                 fid_buf.lb_buf = (void *)mdd_object_fid(tobj);
4077                 fid_buf.lb_len = sizeof(struct lu_fid);
4078                 rc = mdo_declare_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID,
4079                                            0, handle);
4080                 if (rc)
4081                         return rc;
4082
4083                 rc = mdo_declare_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4084                 if (rc)
4085                         return rc;
4086         }
4087
4088         if (!S_ISDIR(attr->la_mode)) {
4089                 rc = mdd_iterate_linkea(env, sobj, tobj, tname,
4090                                         mdd_object_fid(tpobj), ldata, NULL,
4091                                         handle, mdd_declare_update_link);
4092                 if (rc)
4093                         return rc;
4094         }
4095
4096         if (!S_ISDIR(attr->la_mode) || lmv) {
4097                 rc = mdo_declare_ref_del(env, sobj, handle);
4098                 if (rc)
4099                         return rc;
4100
4101                 if (S_ISDIR(attr->la_mode)) {
4102                         rc = mdo_declare_ref_del(env, sobj, handle);
4103                         if (rc)
4104                                 return rc;
4105                 }
4106
4107                 rc = mdo_declare_destroy(env, sobj, handle);
4108                 if (rc)
4109                         return rc;
4110         }
4111
4112         rc = mdd_declare_migrate_update(env, spobj, tpobj, tobj, sname, tname,
4113                                         attr, spattr, tpattr, ldata, ma,
4114                                         handle);
4115         return rc;
4116 }
4117
4118 /**
4119  * migrate dirent from \a spobj to \a tpobj.
4120  **/
4121 static int mdd_migrate_update(const struct lu_env *env,
4122                               struct mdd_object *spobj,
4123                               struct mdd_object *tpobj,
4124                               struct mdd_object *obj,
4125                               const struct lu_name *sname,
4126                               const struct lu_name *tname,
4127                               struct lu_attr *attr,
4128                               struct lu_attr *spattr,
4129                               struct lu_attr *tpattr,
4130                               struct linkea_data *ldata,
4131                               struct md_attr *ma,
4132                               struct thandle *handle)
4133 {
4134         struct mdd_thread_info *info = mdd_env_info(env);
4135         struct lu_attr *la = &info->mdi_la_for_fix;
4136         int rc;
4137
4138         ENTRY;
4139
4140         CDEBUG(D_INFO, "update "DFID" from "DFID"/%s to "DFID"/%s\n",
4141                PFID(mdd_object_fid(obj)), PFID(mdd_object_fid(spobj)),
4142                sname->ln_name, PFID(mdd_object_fid(tpobj)), tname->ln_name);
4143
4144         rc = __mdd_index_delete(env, spobj, sname->ln_name,
4145                                 S_ISDIR(attr->la_mode), handle);
4146         if (rc)
4147                 RETURN(rc);
4148
4149         rc = __mdd_index_insert(env, tpobj, mdd_object_fid(obj),
4150                                 attr->la_mode & S_IFMT,
4151                                 tname->ln_name, handle);
4152         if (rc)
4153                 RETURN(rc);
4154
4155         rc = mdd_links_write(env, obj, ldata, handle);
4156         if (rc)
4157                 RETURN(rc);
4158
4159         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
4160         la->la_valid = LA_CTIME | LA_MTIME;
4161         mdd_write_lock(env, spobj, DT_SRC_PARENT);
4162         rc = mdd_update_time(env, spobj, spattr, la, handle);
4163         mdd_write_unlock(env, spobj);
4164         if (rc)
4165                 RETURN(rc);
4166
4167         if (tpobj != spobj) {
4168                 la->la_valid = LA_CTIME | LA_MTIME;
4169                 mdd_write_lock(env, tpobj, DT_TGT_PARENT);
4170                 rc = mdd_update_time(env, tpobj, tpattr, la, handle);
4171                 mdd_write_unlock(env, tpobj);
4172                 if (rc)
4173                         RETURN(rc);
4174         }
4175
4176         RETURN(rc);
4177 }
4178
4179 /**
4180  * Migrate file/dir to target MDT.
4181  *
4182  * Create target according to \a spec, and then migrate xattrs, if it's
4183  * directory, migrate source stripes to target.
4184  *
4185  * \param[in] env       execution environment
4186  * \param[in] spobj     source parent object
4187  * \param[in] tpobj     target parent object
4188  * \param[in] sobj      source object
4189  * \param[in] tobj      target object
4190  * \param[in] lname     file name
4191  * \param[in] spattr    source parent attributes
4192  * \param[in] tpattr    target parent attributes
4193  * \param[in] attr      source attributes
4194  * \param[in] sbuf      source LMV buf
4195  * \param[in] spec      migrate create spec
4196  * \param[in] hint      target creation hint
4197  * \param[in] handle    tranasction handle
4198  *
4199  * \retval      0 on success
4200  * \retval      -errno on failure
4201  **/
4202 static int mdd_migrate_create(const struct lu_env *env,
4203                               struct mdd_object *spobj,
4204                               struct mdd_object *tpobj,
4205                               struct mdd_object *sobj,
4206                               struct mdd_object *tobj,
4207                               const struct lu_name *sname,
4208                               const struct lu_name *tname,
4209                               struct lu_attr *spattr,
4210                               struct lu_attr *tpattr,
4211                               struct lu_attr *attr,
4212                               const struct lu_buf *sbuf,
4213                               struct linkea_data *ldata,
4214                               struct mdd_xattrs *xattrs,
4215                               struct md_attr *ma,
4216                               struct md_op_spec *spec,
4217                               struct dt_allocation_hint *hint,
4218                               struct thandle *handle)
4219 {
4220         int rc;
4221
4222         ENTRY;
4223
4224         /*
4225          * migrate sobj stripes to tobj if it's directory:
4226          * 1. detach stripes from sobj.
4227          * 2. attach stripes to tobj, see mdd_declare_migrate_mdt().
4228          * 3. create stripes for tobj, see lod_xattr_set_lmv().
4229          */
4230         if (S_ISDIR(attr->la_mode)) {
4231                 struct mdd_thread_info *info = mdd_env_info(env);
4232                 struct md_layout_change *mlc = &info->mdi_mlc;
4233
4234                 mlc->mlc_opc = MD_LAYOUT_DETACH;
4235
4236                 mdd_write_lock(env, sobj, DT_SRC_PARENT);
4237                 rc = mdo_layout_change(env, sobj, mlc, handle);
4238                 mdd_write_unlock(env, sobj);
4239                 if (rc)
4240                         RETURN(rc);
4241         }
4242
4243         /* don't set nlink from sobj */
4244         attr->la_valid &= ~LA_NLINK;
4245
4246         rc = mdd_create_object(env, tpobj, tobj, attr, spec, NULL, NULL, NULL,
4247                                hint, handle, false);
4248         if (rc)
4249                 RETURN(rc);
4250
4251         mdd_write_lock(env, tobj, DT_TGT_CHILD);
4252         rc = mdd_foreach_xattr(env, tobj, xattrs, handle, mdo_xattr_set);
4253         mdd_write_unlock(env, tobj);
4254         if (rc)
4255                 RETURN(rc);
4256
4257         /* for regular file, update OST objects XATTR_NAME_FID */
4258         if (S_ISREG(attr->la_mode)) {
4259                 struct lu_buf fid_buf;
4260
4261                 /* target may be remote, update PFID via sobj. */
4262                 fid_buf.lb_buf = (void *)mdd_object_fid(tobj);
4263                 fid_buf.lb_len = sizeof(struct lu_fid);
4264                 rc = mdo_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID, 0,
4265                                    handle);
4266                 if (rc)
4267                         RETURN(rc);
4268
4269                 /* delete LOV to avoid deleting OST objs when destroying sobj */
4270                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4271                 rc = mdo_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4272                 mdd_write_unlock(env, sobj);
4273                 /* O_DELAY_CREATE file may not have LOV, ignore -ENODATA */
4274                 if (rc && rc != -ENODATA)
4275                         RETURN(rc);
4276                 rc = 0;
4277         }
4278
4279         /* update links FID */
4280         if (!S_ISDIR(attr->la_mode)) {
4281                 rc = mdd_iterate_linkea(env, sobj, tobj, tname,
4282                                         mdd_object_fid(tpobj), ldata,
4283                                         NULL, handle, mdd_update_link);
4284                 if (rc)
4285                         RETURN(rc);
4286         }
4287
4288         /* don't destroy sobj if it's plain directory */
4289         if (!S_ISDIR(attr->la_mode) || sbuf->lb_buf) {
4290                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4291                 rc = mdo_ref_del(env, sobj, handle);
4292                 if (!rc) {
4293                         if (S_ISDIR(attr->la_mode))
4294                                 rc = mdo_ref_del(env, sobj, handle);
4295                         if (!rc)
4296                                 rc = mdo_destroy(env, sobj, handle);
4297                 }
4298                 mdd_write_unlock(env, sobj);
4299                 if (rc)
4300                         RETURN(rc);
4301         }
4302
4303         rc = mdd_migrate_update(env, spobj, tpobj, tobj, sname, tname, attr,
4304                                 spattr, tpattr, ldata, ma, handle);
4305
4306         RETURN(rc);
4307 }
4308
4309 /* NB: if user issued different migrate command, we can't adjust it silently
4310  * here, because this command will decide target MDT in subdir migration in
4311  * LMV.
4312  */
4313 static int mdd_migrate_cmd_check(struct mdd_device *mdd,
4314                                  const struct lmv_mds_md_v1 *lmv,
4315                                  const struct lmv_user_md_v1 *lum,
4316                                  const struct lu_name *lname)
4317 {
4318         __u32 lum_stripe_count = lum->lum_stripe_count;
4319         __u32 lum_hash_type = lum->lum_hash_type &
4320                               cpu_to_le32(LMV_HASH_TYPE_MASK);
4321         __u32 lmv_hash_type = lmv->lmv_hash_type &
4322                               cpu_to_le32(LMV_HASH_TYPE_MASK);
4323
4324         if (!lmv_is_sane(lmv))
4325                 return -EBADF;
4326
4327         /* if stripe_count unspecified, set to 1 */
4328         if (!lum_stripe_count)
4329                 lum_stripe_count = cpu_to_le32(1);
4330
4331         /* TODO: check specific MDTs */
4332         if (lum_stripe_count != lmv->lmv_migrate_offset ||
4333             lum->lum_stripe_offset != lmv->lmv_master_mdt_index ||
4334             (lum_hash_type && lum_hash_type != lmv_hash_type)) {
4335                 CERROR("%s: '"DNAME"' migration was interrupted, run 'lfs migrate -m %d -c %d -H %s "DNAME"' to finish migration.\n",
4336                         mdd2obd_dev(mdd)->obd_name, PNAME(lname),
4337                         le32_to_cpu(lmv->lmv_master_mdt_index),
4338                         le32_to_cpu(lmv->lmv_migrate_offset),
4339                         mdt_hash_name[le32_to_cpu(lmv_hash_type)],
4340                         PNAME(lname));
4341                 return -EPERM;
4342         }
4343
4344         return -EALREADY;
4345 }
4346
4347 /**
4348  * Internal function to migrate directory or file between MDTs.
4349  *
4350  * migrate source to target in following steps:
4351  *   1. create target, append source stripes after target's if it's directory,
4352  *      migrate xattrs and update fid of source links.
4353  *   2. update namespace: migrate dirent from source parent to target parent,
4354  *      update file linkea, and destroy source if it's not needed any more.
4355  *
4356  * \param[in] env       execution environment
4357  * \param[in] spobj     source parent object
4358  * \param[in] tpobj     target parent object
4359  * \param[in] sobj      source object
4360  * \param[in] tobj      target object
4361  * \param[in] sname     source file name
4362  * \param[in] tname     target file name
4363  * \param[in] spec      target creation spec
4364  * \param[in] ma        used to update \a pobj mtime and ctime
4365  *
4366  * \retval              0 on success
4367  * \retval              -errno on failure
4368  */
4369 static int mdd_migrate_object(const struct lu_env *env,
4370                               struct mdd_object *spobj,
4371                               struct mdd_object *tpobj,
4372                               struct mdd_object *sobj,
4373                               struct mdd_object *tobj,
4374                               const struct lu_name *sname,
4375                               const struct lu_name *tname,
4376                               struct md_op_spec *spec,
4377                               struct md_attr *ma)
4378 {
4379         struct mdd_thread_info *info = mdd_env_info(env);
4380         struct mdd_device *mdd = mdo2mdd(&spobj->mod_obj);
4381         struct lu_attr *spattr = &info->mdi_pattr;
4382         struct lu_attr *tpattr = &info->mdi_tpattr;
4383         struct lu_attr *attr = &info->mdi_cattr;
4384         struct linkea_data *ldata = &info->mdi_link_data;
4385         struct dt_allocation_hint *hint = &info->mdi_hint;
4386         struct lu_buf sbuf = { NULL };
4387         struct mdd_xattrs xattrs;
4388         struct lmv_mds_md_v1 *lmv;
4389         struct thandle *handle;
4390         int rc;
4391
4392         ENTRY;
4393
4394         CDEBUG(D_INFO, "migrate %s from "DFID"/"DFID" to "DFID"/"DFID"\n",
4395                sname->ln_name, PFID(mdd_object_fid(spobj)),
4396                PFID(mdd_object_fid(sobj)), PFID(mdd_object_fid(tpobj)),
4397                PFID(mdd_object_fid(tobj)));
4398
4399         rc = mdd_la_get(env, sobj, attr);
4400         if (rc)
4401                 RETURN(rc);
4402
4403         rc = mdd_la_get(env, spobj, spattr);
4404         if (rc)
4405                 RETURN(rc);
4406
4407         rc = mdd_la_get(env, tpobj, tpattr);
4408         if (rc)
4409                 RETURN(rc);
4410
4411         rc = mdd_migrate_sanity_check(env, mdd, spobj, tpobj, sobj, tobj,
4412                                       spattr, tpattr, attr);
4413         if (rc)
4414                 RETURN(rc);
4415
4416         mdd_xattrs_init(&xattrs);
4417
4418         if (S_ISDIR(attr->la_mode) && !spec->sp_migrate_nsonly) {
4419                 struct lmv_user_md_v1 *lum = spec->u.sp_ea.eadata;
4420
4421                 LASSERT(lum);
4422
4423                 /* if user use default value '0' for stripe_count, we need to
4424                  * adjust it to '1' to create a 1-stripe directory.
4425                  */
4426                 if (lum->lum_stripe_count == 0)
4427                         lum->lum_stripe_count = cpu_to_le32(1);
4428
4429                 rc = mdd_stripe_get(env, sobj, &sbuf, XATTR_NAME_LMV);
4430                 if (rc && rc != -ENODATA)
4431                         GOTO(out, rc);
4432
4433                 lmv = sbuf.lb_buf;
4434                 if (lmv) {
4435                         if (!lmv_is_sane(lmv))
4436                                 GOTO(out, rc = -EBADF);
4437                         if (lmv_is_migrating(lmv)) {
4438                                 rc = mdd_migrate_cmd_check(mdd, lmv, lum,
4439                                                            sname);
4440                                 GOTO(out, rc);
4441                         }
4442                 }
4443         } else if (!S_ISDIR(attr->la_mode)) {
4444                 if (spobj == tpobj)
4445                         GOTO(out, rc = -EALREADY);
4446
4447                 /* update namespace only if @sobj is on MDT where @tpobj is. */
4448                 if (!mdd_object_remote(tpobj) && !mdd_object_remote(sobj))
4449                         spec->sp_migrate_nsonly = true;
4450
4451                 if (S_ISLNK(attr->la_mode)) {
4452                         lu_buf_check_and_alloc(&sbuf, attr->la_size + 1);
4453                         if (!sbuf.lb_buf)
4454                                 GOTO(out, rc = -ENOMEM);
4455
4456                         rc = mdd_readlink(env, &sobj->mod_obj, &sbuf);
4457                         if (rc <= 0) {
4458                                 rc = rc ?: -EFAULT;
4459                                 CERROR("%s: "DFID" readlink failed: rc = %d\n",
4460                                        mdd2obd_dev(mdd)->obd_name,
4461                                        PFID(mdd_object_fid(sobj)), rc);
4462                                 GOTO(out, rc);
4463                         }
4464                 }
4465         }
4466
4467         /* linkea needs update upon FID or parent stripe change */
4468         rc = mdd_migrate_linkea_prepare(env, mdd, spobj, tpobj, sobj, sname,
4469                                         tname, attr, ldata);
4470         if (rc > 0)
4471                 /* update namespace only if @sobj has link on its MDT. */
4472                 spec->sp_migrate_nsonly = true;
4473         else if (rc < 0)
4474                 GOTO(out, rc);
4475
4476         /* migrate inode will migrate xattrs, prepare xattrs early to avoid
4477          * RPCs inside transaction.
4478          */
4479         if (!spec->sp_migrate_nsonly) {
4480                 rc = mdd_xattrs_migrate_prep(env, &xattrs, sobj, true, true);
4481                 if (rc)
4482                         GOTO(out, rc);
4483         }
4484
4485         handle = mdd_trans_create(env, mdd);
4486         if (IS_ERR(handle))
4487                 GOTO(out, rc = PTR_ERR(handle));
4488
4489         if (spec->sp_migrate_nsonly)
4490                 rc = mdd_declare_migrate_update(env, spobj, tpobj, sobj, sname,
4491                                                 tname, attr, spattr, tpattr,
4492                                                 ldata, ma, handle);
4493         else
4494                 rc = mdd_declare_migrate_create(env, spobj, tpobj, sobj, tobj,
4495                                                 sname, tname, spattr, tpattr,
4496                                                 attr, &sbuf, ldata, &xattrs, ma,
4497                                                 spec, hint, handle);
4498         if (rc)
4499                 GOTO(stop, rc);
4500
4501         rc = mdd_declare_changelog_store(env, mdd, CL_MIGRATE, tname, sname,
4502                                          handle);
4503         if (rc)
4504                 GOTO(stop, rc);
4505
4506         rc = mdd_trans_start(env, mdd, handle);
4507         if (rc)
4508                 GOTO(stop, rc);
4509
4510         if (spec->sp_migrate_nsonly)
4511                 rc = mdd_migrate_update(env, spobj, tpobj, sobj, sname, tname,
4512                                         attr, spattr, tpattr, ldata, ma,
4513                                         handle);
4514         else
4515                 rc = mdd_migrate_create(env, spobj, tpobj, sobj, tobj, sname,
4516                                         tname, spattr, tpattr, attr, &sbuf,
4517                                         ldata, &xattrs, ma, spec, hint, handle);
4518         if (rc)
4519                 GOTO(stop, rc);
4520
4521         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0,
4522                                     spec->sp_migrate_nsonly ? sobj : tobj,
4523                                     mdd_object_fid(spobj), mdd_object_fid(sobj),
4524                                     mdd_object_fid(tpobj), tname, sname,
4525                                     handle);
4526         if (rc)
4527                 GOTO(stop, rc);
4528         EXIT;
4529
4530 stop:
4531         rc = mdd_trans_stop(env, mdd, rc, handle);
4532 out:
4533         mdd_xattrs_fini(&xattrs);
4534         lu_buf_free(&sbuf);
4535
4536         return rc;
4537 }
4538
4539 /**
4540  * Migrate directory or file between MDTs.
4541  *
4542  * \param[in] env       execution environment
4543  * \param[in] md_spobj  source parent object
4544  * \param[in] md_tpobj  target parent object
4545  * \param[in] md_sobj   source object
4546  * \param[in] lname     file name
4547  * \param[in] md_tobj   target object
4548  * \param[in] spec      target creation spec
4549  * \param[in] ma        used to update \a pobj mtime and ctime
4550  *
4551  * \retval              0 on success
4552  * \retval              -errno on failure
4553  */
4554 static int mdd_migrate(const struct lu_env *env, struct md_object *md_spobj,
4555                        struct md_object *md_tpobj, struct md_object *md_sobj,
4556                        struct md_object *md_tobj, const struct lu_name *lname,
4557                        struct md_op_spec *spec, struct md_attr *ma)
4558 {
4559         return mdd_migrate_object(env, md2mdd_obj(md_spobj),
4560                                   md2mdd_obj(md_tpobj), md2mdd_obj(md_sobj),
4561                                   md2mdd_obj(md_tobj), lname, lname, spec, ma);
4562 }
4563
4564 static int mdd_declare_1sd_collapse(const struct lu_env *env,
4565                                     struct mdd_object *pobj,
4566                                     struct mdd_object *obj,
4567                                     struct mdd_object *stripe,
4568                                     struct lu_attr *attr,
4569                                     struct mdd_xattrs *xattrs,
4570                                     struct md_layout_change *mlc,
4571                                     struct lu_name *lname,
4572                                     struct thandle *handle)
4573 {
4574         int rc;
4575
4576         mlc->mlc_opc = MD_LAYOUT_DETACH;
4577         rc = mdo_declare_layout_change(env, obj, mlc, handle);
4578         if (rc)
4579                 return rc;
4580
4581         rc = mdo_declare_index_insert(env, stripe, mdd_object_fid(pobj),
4582                                       S_IFDIR, dotdot, handle);
4583         if (rc)
4584                 return rc;
4585
4586         rc = mdd_foreach_xattr(env, stripe, xattrs, handle,
4587                                mdo_declare_xattr_set);
4588         if (rc)
4589                 return rc;
4590
4591         rc = mdo_declare_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4592         if (rc)
4593                 return rc;
4594
4595         rc = mdo_declare_attr_set(env, stripe, attr, handle);
4596         if (rc)
4597                 return rc;
4598
4599         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
4600         if (rc)
4601                 return rc;
4602
4603         rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(stripe),
4604                                       attr->la_mode, lname->ln_name, handle);
4605         if (rc)
4606                 return rc;
4607
4608         rc = mdo_declare_ref_del(env, obj, handle);
4609         if (rc)
4610                 return rc;
4611
4612         rc = mdo_declare_ref_del(env, obj, handle);
4613         if (rc)
4614                 return rc;
4615
4616         rc = mdo_declare_destroy(env, obj, handle);
4617         if (rc)
4618                 return rc;
4619
4620         return rc;
4621 }
4622
4623 /* transform one-stripe directory to a plain directory */
4624 static int mdd_1sd_collapse(const struct lu_env *env,
4625                             struct mdd_object *pobj,
4626                             struct mdd_object *obj,
4627                             struct mdd_object *stripe,
4628                             struct lu_attr *attr,
4629                             struct mdd_xattrs *xattrs,
4630                             struct md_layout_change *mlc,
4631                             struct lu_name *lname,
4632                             struct thandle *handle)
4633 {
4634         int rc;
4635
4636         ENTRY;
4637
4638         /* replace 1-stripe directory with its stripe */
4639         mlc->mlc_opc = MD_LAYOUT_DETACH;
4640
4641         mdd_write_lock(env, obj, DT_SRC_PARENT);
4642         rc = mdo_layout_change(env, obj, mlc, handle);
4643         mdd_write_unlock(env, obj);
4644         if (rc)
4645                 RETURN(rc);
4646
4647         mdd_write_lock(env, pobj, DT_SRC_PARENT);
4648         mdd_write_lock(env, obj, DT_SRC_CHILD);
4649
4650         /* insert dotdot to stripe which points to parent */
4651         rc = __mdd_index_insert_only(env, stripe, mdd_object_fid(pobj),
4652                                      S_IFDIR, dotdot, handle);
4653         if (rc)
4654                 GOTO(out, rc);
4655
4656         rc = mdd_foreach_xattr(env, stripe, xattrs, handle, mdo_xattr_set);
4657         if (rc)
4658                 GOTO(out, rc);
4659
4660         /* delete LMV */
4661         rc = mdo_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4662         if (rc)
4663                 GOTO(out, rc);
4664
4665         /* don't set nlink from parent */
4666         attr->la_valid &= ~LA_NLINK;
4667
4668         rc = mdo_attr_set(env, stripe, attr, handle);
4669         if (rc)
4670                 GOTO(out, rc);
4671
4672         /* delete dir name from parent */
4673         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
4674         if (rc)
4675                 GOTO(out, rc);
4676
4677         /* insert stripe to parent with dir name */
4678         rc = __mdd_index_insert_only(env, pobj, mdd_object_fid(stripe),
4679                                      attr->la_mode, lname->ln_name, handle);
4680         if (rc)
4681                 GOTO(out, rc);
4682
4683         /* destroy dir obj */
4684         rc = mdo_ref_del(env, obj, handle);
4685         if (rc)
4686                 GOTO(out, rc);
4687
4688         rc = mdo_ref_del(env, obj, handle);
4689         if (rc)
4690                 GOTO(out, rc);
4691
4692         rc = mdo_destroy(env, obj, handle);
4693         if (rc)
4694                 GOTO(out, rc);
4695
4696         EXIT;
4697 out:
4698         mdd_write_unlock(env, obj);
4699         mdd_write_unlock(env, pobj);
4700
4701         return rc;
4702 }
4703
4704 /*
4705  * shrink directory stripes after migration/merge
4706  */
4707 int mdd_dir_layout_shrink(const struct lu_env *env,
4708                           struct md_object *md_obj,
4709                           struct md_layout_change *mlc)
4710 {
4711         struct mdd_device *mdd = mdo2mdd(md_obj);
4712         struct mdd_thread_info *info = mdd_env_info(env);
4713         struct mdd_object *obj = md2mdd_obj(md_obj);
4714         struct mdd_object *pobj = NULL;
4715         struct mdd_object *stripe = NULL;
4716         struct lu_attr *attr = &info->mdi_pattr;
4717         struct lu_fid *fid = &info->mdi_fid2;
4718         struct lu_name lname = { NULL };
4719         struct lu_buf lmv_buf = { NULL };
4720         struct mdd_xattrs xattrs;
4721         struct lmv_mds_md_v1 *lmv;
4722         struct lmv_user_md *lmu;
4723         struct thandle *handle;
4724         int rc;
4725
4726         ENTRY;
4727
4728         rc = mdd_la_get(env, obj, attr);
4729         if (rc)
4730                 RETURN(rc);
4731
4732         if (!S_ISDIR(attr->la_mode))
4733                 RETURN(-ENOTDIR);
4734
4735         rc = mdd_stripe_get(env, obj, &lmv_buf, XATTR_NAME_LMV);
4736         if (rc < 0)
4737                 RETURN(rc);
4738
4739         lmv = lmv_buf.lb_buf;
4740         if (!lmv_is_sane(lmv))
4741                 RETURN(-EBADF);
4742
4743         lmu = mlc->mlc_buf.lb_buf;
4744
4745         /* adjust the default value '0' to '1' */
4746         if (lmu->lum_stripe_count == 0)
4747                 lmu->lum_stripe_count = cpu_to_le32(1);
4748
4749         /* these were checked in MDT */
4750         LASSERT(le32_to_cpu(lmu->lum_stripe_count) <
4751                 le32_to_cpu(lmv->lmv_stripe_count));
4752         LASSERT(!lmv_is_splitting(lmv));
4753         LASSERT(lmv_is_migrating(lmv) || lmv_is_merging(lmv));
4754
4755         mdd_xattrs_init(&xattrs);
4756
4757         /* if dir stripe count will be shrunk to 1, it needs to be transformed
4758          * to a plain dir, which will cause FID change and namespace update.
4759          */
4760         if (le32_to_cpu(lmu->lum_stripe_count) == 1) {
4761                 struct linkea_data *ldata = &info->mdi_link_data;
4762                 char *filename = info->mdi_name;
4763
4764                 rc = mdd_links_read(env, obj, ldata);
4765                 if (rc)
4766                         GOTO(out, rc);
4767
4768                 if (ldata->ld_leh->leh_reccount > 1)
4769                         GOTO(out, rc = -EINVAL);
4770
4771                 linkea_first_entry(ldata);
4772                 if (!ldata->ld_lee)
4773                         GOTO(out, rc = -ENODATA);
4774
4775                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
4776                                     fid);
4777
4778                 /* Note: lname might miss \0 at the end */
4779                 snprintf(filename, sizeof(info->mdi_name), "%.*s",
4780                          lname.ln_namelen, lname.ln_name);
4781                 lname.ln_name = filename;
4782
4783                 pobj = mdd_object_find(env, mdd, fid);
4784                 if (IS_ERR(pobj)) {
4785                         rc = PTR_ERR(pobj);
4786                         pobj = NULL;
4787                         GOTO(out, rc);
4788                 }
4789
4790                 fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[0]);
4791
4792                 stripe = mdd_object_find(env, mdd, fid);
4793                 if (IS_ERR(stripe)) {
4794                         mdd_object_put(env, pobj);
4795                         pobj = NULL;
4796                         GOTO(out, rc = PTR_ERR(stripe));
4797                 }
4798
4799                 if (!lmv_is_fixed(lmv))
4800                         rc = mdd_xattrs_migrate_prep(env, &xattrs, obj, false,
4801                                                      false);
4802         }
4803
4804         handle = mdd_trans_create(env, mdd);
4805         if (IS_ERR(handle))
4806                 GOTO(out, rc = PTR_ERR(handle));
4807
4808         mlc->mlc_opc = MD_LAYOUT_SHRINK;
4809         rc = mdo_declare_layout_change(env, obj, mlc, handle);
4810         if (rc)
4811                 GOTO(stop_trans, rc);
4812
4813         if (le32_to_cpu(lmu->lum_stripe_count) == 1 && !lmv_is_fixed(lmv)) {
4814                 rc = mdd_declare_1sd_collapse(env, pobj, obj, stripe, attr,
4815                                               &xattrs, mlc, &lname, handle);
4816                 if (rc)
4817                         GOTO(stop_trans, rc);
4818         }
4819
4820         rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL, NULL,
4821                                          handle);
4822         if (rc)
4823                 GOTO(stop_trans, rc);
4824
4825         rc = mdd_trans_start(env, mdd, handle);
4826         if (rc)
4827                 GOTO(stop_trans, rc);
4828
4829         mdd_write_lock(env, obj, DT_SRC_PARENT);
4830         mlc->mlc_opc = MD_LAYOUT_SHRINK;
4831         rc = mdo_layout_change(env, obj, mlc, handle);
4832         mdd_write_unlock(env, obj);
4833         if (rc)
4834                 GOTO(stop_trans, rc);
4835
4836         if (le32_to_cpu(lmu->lum_stripe_count) == 1 && !lmv_is_fixed(lmv)) {
4837                 rc = mdd_1sd_collapse(env, pobj, obj, stripe, attr, &xattrs,
4838                                       mlc, &lname, handle);
4839                 if (rc)
4840                         GOTO(stop_trans, rc);
4841         }
4842
4843         rc = mdd_changelog_data_store_xattr(env, mdd, CL_LAYOUT, 0, obj,
4844                                             XATTR_NAME_LMV, handle);
4845         GOTO(stop_trans, rc);
4846
4847 stop_trans:
4848         rc = mdd_trans_stop(env, mdd, rc, handle);
4849 out:
4850         mdd_xattrs_fini(&xattrs);
4851         if (pobj) {
4852                 mdd_object_put(env, stripe);
4853                 mdd_object_put(env, pobj);
4854         }
4855         lu_buf_free(&lmv_buf);
4856         return rc;
4857 }
4858
4859 static int mdd_dir_declare_split_plain(const struct lu_env *env,
4860                                         struct mdd_device *mdd,
4861                                         struct mdd_object *pobj,
4862                                         struct mdd_object *obj,
4863                                         struct mdd_object *tobj,
4864                                         struct mdd_xattrs *xattrs,
4865                                         struct md_layout_change *mlc,
4866                                         struct dt_allocation_hint *hint,
4867                                         struct thandle *handle)
4868 {
4869         struct mdd_thread_info *info = mdd_env_info(env);
4870         const struct lu_name *lname = mlc->mlc_name;
4871         struct lu_attr *la = &info->mdi_la_for_fix;
4872         struct lmv_user_md_v1 *lum = mlc->mlc_spec->u.sp_ea.eadata;
4873         struct linkea_data *ldata = &info->mdi_link_data;
4874         struct lmv_mds_md_v1 *lmv;
4875         __u32 count;
4876         int rc;
4877
4878         mlc->mlc_opc = MD_LAYOUT_DETACH;
4879         rc = mdo_declare_layout_change(env, obj, mlc, handle);
4880         if (rc)
4881                 return rc;
4882
4883         memset(ldata, 0, sizeof(*ldata));
4884         rc = mdd_linkea_prepare(env, obj, NULL, NULL, mdd_object_fid(pobj),
4885                                 lname, 1, 0, ldata);
4886         if (rc)
4887                 return rc;
4888
4889         count = lum->lum_stripe_count;
4890         lum->lum_stripe_count = 0;
4891         /* don't set default LMV since it will become a striped dir  */
4892         lum->lum_max_inherit = LMV_INHERIT_NONE;
4893         mdd_object_make_hint(env, pobj, tobj, mlc->mlc_attr, mlc->mlc_spec,
4894                              hint);
4895         rc = mdd_declare_create(env, mdo2mdd(&pobj->mod_obj), pobj, tobj,
4896                                 lname, mlc->mlc_attr, handle, mlc->mlc_spec,
4897                                 ldata, NULL, NULL, NULL, hint);
4898         if (rc)
4899                 return rc;
4900
4901         /* tobj mode will be used in lod_declare_xattr_set(), but it's not
4902          * created yet.
4903          */
4904         tobj->mod_obj.mo_lu.lo_header->loh_attr |= S_IFDIR;
4905
4906         lmv = (typeof(lmv))info->mdi_key;
4907         memset(lmv, 0, sizeof(*lmv));
4908         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
4909         lmv->lmv_stripe_count = cpu_to_le32(1);
4910         lmv->lmv_hash_type = cpu_to_le32(LMV_HASH_TYPE_DEFAULT);
4911         fid_le_to_cpu(&lmv->lmv_stripe_fids[0], mdd_object_fid(obj));
4912
4913         mlc->mlc_opc = MD_LAYOUT_ATTACH;
4914         mlc->mlc_buf.lb_buf = lmv;
4915         mlc->mlc_buf.lb_len = lmv_mds_md_size(1, LMV_MAGIC_V1);
4916         rc = mdo_declare_layout_change(env, tobj, mlc, handle);
4917         if (rc)
4918                 return rc;
4919
4920         rc = mdd_foreach_xattr(env, tobj, xattrs, handle,
4921                                mdo_declare_xattr_set);
4922         if (rc)
4923                 return rc;
4924
4925         lum->lum_stripe_count = count;
4926         mlc->mlc_opc = MD_LAYOUT_SPLIT;
4927         rc = mdo_declare_layout_change(env, tobj, mlc, handle);
4928         if (rc)
4929                 return rc;
4930
4931         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
4932         if (rc)
4933                 return rc;
4934
4935         rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(tobj),
4936                                       S_IFDIR, lname->ln_name, handle);
4937         if (rc)
4938                 return rc;
4939
4940         la->la_valid = LA_CTIME | LA_MTIME;
4941         rc = mdo_declare_attr_set(env, obj, la, handle);
4942         if (rc)
4943                 return rc;
4944
4945         rc = mdo_declare_attr_set(env, pobj, la, handle);
4946         if (rc)
4947                 return rc;
4948
4949         rc = mdd_declare_changelog_store(env, mdd, CL_MIGRATE, lname, NULL,
4950                                          handle);
4951         return rc;
4952 }
4953
4954 /**
4955  * plain directory split:
4956  * 1. create \a tobj as plain directory.
4957  * 2. append \a obj as first stripe of \a tobj.
4958  * 3. migrate xattrs from \a obj to \a tobj.
4959  * 4. split \a tobj to specific stripe count.
4960  */
4961 static int mdd_dir_split_plain(const struct lu_env *env,
4962                                 struct mdd_device *mdd,
4963                                 struct mdd_object *pobj,
4964                                 struct mdd_object *obj,
4965                                 struct mdd_object *tobj,
4966                                 struct mdd_xattrs *xattrs,
4967                                 struct md_layout_change *mlc,
4968                                 struct dt_allocation_hint *hint,
4969                                 struct thandle *handle)
4970 {
4971         struct mdd_thread_info *info = mdd_env_info(env);
4972         struct lu_attr *pattr = &info->mdi_pattr;
4973         struct lu_attr *la = &info->mdi_la_for_fix;
4974         const struct lu_name *lname = mlc->mlc_name;
4975         struct linkea_data *ldata = &info->mdi_link_data;
4976         int rc;
4977
4978         ENTRY;
4979
4980         /* copy linkea out and set on target later */
4981         rc = mdd_links_read(env, obj, ldata);
4982         if (rc)
4983                 RETURN(rc);
4984
4985         mlc->mlc_opc = MD_LAYOUT_DETACH;
4986         rc = mdo_layout_change(env, obj, mlc, handle);
4987         if (rc)
4988                 RETURN(rc);
4989
4990         /* don't set nlink from obj */
4991         mlc->mlc_attr->la_valid &= ~LA_NLINK;
4992
4993         rc = mdd_create_object(env, pobj, tobj, mlc->mlc_attr, mlc->mlc_spec,
4994                                NULL, NULL, NULL, hint, handle, false);
4995         if (rc)
4996                 RETURN(rc);
4997
4998         rc = mdd_foreach_xattr(env, tobj, xattrs, handle, mdo_xattr_set);
4999         if (rc)
5000                 RETURN(rc);
5001
5002         rc = mdd_links_write(env, tobj, ldata, handle);
5003         if (rc)
5004                 RETURN(rc);
5005
5006         rc = __mdd_index_delete(env, pobj, lname->ln_name, true, handle);
5007         if (rc)
5008                 RETURN(rc);
5009
5010         rc = __mdd_index_insert(env, pobj, mdd_object_fid(tobj), S_IFDIR,
5011                                 lname->ln_name, handle);
5012         if (rc)
5013                 RETURN(rc);
5014
5015         la->la_ctime = la->la_mtime = mlc->mlc_attr->la_mtime;
5016         la->la_valid = LA_CTIME | LA_MTIME;
5017
5018         mdd_write_lock(env, obj, DT_SRC_CHILD);
5019         rc = mdd_update_time(env, tobj, mlc->mlc_attr, la, handle);
5020         mdd_write_unlock(env, obj);
5021         if (rc)
5022                 RETURN(rc);
5023
5024         rc = mdd_la_get(env, pobj, pattr);
5025         if (rc)
5026                 RETURN(rc);
5027
5028         la->la_valid = LA_CTIME | LA_MTIME;
5029
5030         mdd_write_lock(env, pobj, DT_SRC_PARENT);
5031         rc = mdd_update_time(env, pobj, pattr, la, handle);
5032         mdd_write_unlock(env, pobj);
5033         if (rc)
5034                 RETURN(rc);
5035
5036         /* FID changes, record it as CL_MIGRATE */
5037         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0, tobj,
5038                                     mdd_object_fid(pobj), mdd_object_fid(obj),
5039                                     mdd_object_fid(pobj), lname, lname, handle);
5040         RETURN(rc);
5041 }
5042
5043 int mdd_dir_layout_split(const struct lu_env *env, struct md_object *o,
5044                          struct md_layout_change *mlc)
5045 {
5046         struct mdd_thread_info *info = mdd_env_info(env);
5047         struct mdd_device *mdd = mdo2mdd(o);
5048         struct mdd_object *obj = md2mdd_obj(o);
5049         struct mdd_object *pobj = md2mdd_obj(mlc->mlc_parent);
5050         struct mdd_object *tobj = md2mdd_obj(mlc->mlc_target);
5051         struct dt_allocation_hint *hint = &info->mdi_hint;
5052         bool is_plain = false;
5053         struct mdd_xattrs xattrs;
5054         struct thandle *handle;
5055         int rc;
5056
5057         ENTRY;
5058
5059         LASSERT(S_ISDIR(mdd_object_type(obj)));
5060
5061         rc = mdo_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LMV);
5062         if (rc == -ENODATA)
5063                 is_plain = true;
5064         else if (rc < 0)
5065                 RETURN(rc);
5066
5067         mdd_xattrs_init(&xattrs);
5068         if (is_plain)
5069                 rc = mdd_xattrs_migrate_prep(env, &xattrs, obj, true, true);
5070
5071         handle = mdd_trans_create(env, mdd);
5072         if (IS_ERR(handle))
5073                 GOTO(out, rc = PTR_ERR(handle));
5074
5075         if (is_plain) {
5076                 rc = mdd_dir_declare_split_plain(env, mdd, pobj, obj, tobj,
5077                                                  &xattrs, mlc, hint, handle);
5078         } else {
5079                 mlc->mlc_opc = MD_LAYOUT_SPLIT;
5080                 rc = mdo_declare_layout_change(env, obj, mlc, handle);
5081                 if (rc)
5082                         GOTO(stop_trans, rc);
5083
5084                 rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL,
5085                                                  NULL, handle);
5086         }
5087         if (rc)
5088                 GOTO(stop_trans, rc);
5089
5090         rc = mdd_trans_start(env, mdd, handle);
5091         if (rc)
5092                 GOTO(stop_trans, rc);
5093
5094         if (is_plain) {
5095                 rc = mdd_dir_split_plain(env, mdd, pobj, obj, tobj, &xattrs,
5096                                          mlc, hint, handle);
5097         } else {
5098                 mdd_write_lock(env, obj, DT_TGT_CHILD);
5099                 rc = mdo_xattr_set(env, obj, NULL, XATTR_NAME_LMV,
5100                                    LU_XATTR_CREATE, handle);
5101                 mdd_write_unlock(env, obj);
5102                 if (rc)
5103                         GOTO(stop_trans, rc);
5104
5105                 rc = mdd_changelog_data_store_xattr(env, mdd, CL_LAYOUT, 0, obj,
5106                                                     XATTR_NAME_LMV, handle);
5107         }
5108         if (rc)
5109                 GOTO(stop_trans, rc);
5110
5111         EXIT;
5112
5113 stop_trans:
5114         rc = mdd_trans_stop(env, mdd, rc, handle);
5115 out:
5116         mdd_xattrs_fini(&xattrs);
5117
5118         return rc;
5119 }
5120
5121 const struct md_dir_operations mdd_dir_ops = {
5122         .mdo_is_subdir     = mdd_is_subdir,
5123         .mdo_lookup        = mdd_lookup,
5124         .mdo_create        = mdd_create,
5125         .mdo_rename        = mdd_rename,
5126         .mdo_link          = mdd_link,
5127         .mdo_unlink        = mdd_unlink,
5128         .mdo_create_data   = mdd_create_data,
5129         .mdo_migrate       = mdd_migrate,
5130 };