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