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         strscpy(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         strscpy(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         strscpy(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,
3536                                                  mdd_object_fid(mdd_tobj),
3537                                                  mdd_object_type(mdd_tobj),
3538                                                  tname, handle);
3539                         if (rc2)
3540                                 CWARN("%s: tpobj "DFID" fix error: rc = %d\n",
3541                                       mdd2obd_dev(mdd)->obd_name,
3542                                       PFID(tpobj_fid), rc2);
3543                 }
3544         }
3545
3546 fixup_spobj:
3547         if (rc && is_dir && mdd_sobj && mdd_spobj != mdd_tpobj) {
3548                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle);
3549                 if (rc2)
3550                         CWARN("%s: spobj "DFID" dotdot delete error: rc = %d\n",
3551                               mdd2obd_dev(mdd)->obd_name, PFID(spobj_fid), rc2);
3552
3553
3554                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid, S_IFDIR,
3555                                               dotdot, handle);
3556                 if (rc2)
3557                         CWARN("%s: spobj "DFID" dotdot insert error: rc = %d\n",
3558                               mdd2obd_dev(mdd)->obd_name, PFID(spobj_fid), rc2);
3559         }
3560
3561 fixup_spobj2:
3562         if (rc) {
3563                 rc2 = __mdd_index_insert(env, mdd_spobj, lf,
3564                                          mdd_object_type(mdd_sobj), sname,
3565                                          handle);
3566                 if (rc2)
3567                         CWARN("%s: spobj "DFID" fix error: rc = %d\n",
3568                               mdd2obd_dev(mdd)->obd_name, PFID(spobj_fid), rc2);
3569         }
3570
3571 cleanup:
3572         if (tobj_locked)
3573                 mdd_write_unlock(env, mdd_tobj);
3574
3575         if (rc == 0)
3576                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, cl_flags,
3577                                             mdd_tobj, mdd_tpobj, tpattr, lf,
3578                                             mdd_spobj, spattr, ltname, lsname,
3579                                             handle);
3580
3581 stop:
3582         rc = mdd_trans_stop(env, mdd, rc, handle);
3583
3584 out_pending:
3585         mdd_object_put(env, mdd_sobj);
3586
3587         return rc;
3588 }
3589
3590 /**
3591  * Check whether we should migrate the file/dir
3592  * return val
3593  *      < 0  permission check failed or other error.
3594  *      = 0  the file can be migrated.
3595  **/
3596 static int mdd_migrate_sanity_check(const struct lu_env *env,
3597                                     struct mdd_device *mdd,
3598                                     struct mdd_object *spobj,
3599                                     struct mdd_object *tpobj,
3600                                     struct mdd_object *sobj,
3601                                     struct mdd_object *tobj,
3602                                     const struct lu_attr *spattr,
3603                                     const struct lu_attr *tpattr,
3604                                     const struct lu_attr *attr)
3605 {
3606         int rc;
3607
3608         ENTRY;
3609
3610         if (!mdd_object_remote(sobj)) {
3611                 mdd_read_lock(env, sobj, DT_SRC_CHILD);
3612                 if (sobj->mod_count > 0) {
3613                         CDEBUG(D_INFO, "%s: "DFID" is opened, count %d\n",
3614                                mdd_obj_dev_name(sobj),
3615                                PFID(mdd_object_fid(sobj)),
3616                                sobj->mod_count);
3617                         mdd_read_unlock(env, sobj);
3618                         RETURN(-EBUSY);
3619                 }
3620                 mdd_read_unlock(env, sobj);
3621         }
3622
3623         if (mdd_object_exists(tobj))
3624                 RETURN(-EEXIST);
3625
3626         rc = mdd_may_delete(env, spobj, spattr, sobj, attr, NULL, 1, 0);
3627         if (rc)
3628                 RETURN(rc);
3629
3630         rc = mdd_may_create(env, tpobj, tpattr, NULL, true);
3631
3632         RETURN(rc);
3633 }
3634
3635 struct mdd_xattr_entry {
3636         struct list_head        mxe_linkage;
3637         char                   *mxe_name;
3638         struct lu_buf           mxe_buf;
3639 };
3640
3641 struct mdd_xattrs {
3642         struct lu_buf           mx_namebuf;
3643         struct list_head        mx_list;
3644 };
3645
3646 static inline void mdd_xattrs_init(struct mdd_xattrs *xattrs)
3647 {
3648         INIT_LIST_HEAD(&xattrs->mx_list);
3649         xattrs->mx_namebuf.lb_buf = NULL;
3650         xattrs->mx_namebuf.lb_len = 0;
3651 }
3652
3653 static inline void mdd_xattrs_fini(struct mdd_xattrs *xattrs)
3654 {
3655         struct mdd_xattr_entry *entry;
3656         struct mdd_xattr_entry *tmp;
3657
3658         list_for_each_entry_safe(entry, tmp, &xattrs->mx_list, mxe_linkage) {
3659                 lu_buf_free(&entry->mxe_buf);
3660                 list_del(&entry->mxe_linkage);
3661                 OBD_FREE_PTR(entry);
3662         }
3663
3664         lu_buf_free(&xattrs->mx_namebuf);
3665 }
3666
3667 /* read xattrs into buf, but skip LMA, LMV, LINKEA if 'skip_linkea' is
3668  * set, and DMV if 'skip_dmv" is set.
3669  */
3670 static int mdd_xattrs_migrate_prep(const struct lu_env *env,
3671                                    struct mdd_xattrs *xattrs,
3672                                    struct mdd_object *sobj,
3673                                    bool skip_linkea,
3674                                    bool skip_dmv)
3675 {
3676         struct lu_attr *attr = MDD_ENV_VAR(env, cattr);
3677         struct mdd_xattr_entry *entry;
3678         bool needencxattr = false;
3679         bool encxattrfound = false;
3680         char *xname;
3681         int list_xsize;
3682         int xlen;
3683         int rem;
3684         int xsize;
3685         int rc;
3686
3687         ENTRY;
3688
3689         list_xsize = mdo_xattr_list(env, sobj, &LU_BUF_NULL);
3690         if (list_xsize == -ENODATA)
3691                 RETURN(0);
3692
3693         if (list_xsize < 0)
3694                 RETURN(list_xsize);
3695
3696         if (attr->la_valid & LA_FLAGS &&
3697             attr->la_flags & LUSTRE_ENCRYPT_FL) {
3698                 needencxattr = true;
3699                 list_xsize +=
3700                         strlen(LL_XATTR_NAME_ENCRYPTION_CONTEXT) + 1;
3701         }
3702
3703         lu_buf_alloc(&xattrs->mx_namebuf, list_xsize);
3704         if (xattrs->mx_namebuf.lb_buf == NULL)
3705                 RETURN(-ENOMEM);
3706
3707         rc = mdo_xattr_list(env, sobj, &xattrs->mx_namebuf);
3708         if (rc < 0)
3709                 GOTO(fini, rc);
3710
3711         rem = rc;
3712         rc = 0;
3713         xname = xattrs->mx_namebuf.lb_buf;
3714 reloop:
3715         for (; rem > 0; xname += xlen, rem -= xlen) {
3716                 if (needencxattr &&
3717                     strcmp(xname, LL_XATTR_NAME_ENCRYPTION_CONTEXT) == 0)
3718                         encxattrfound = true;
3719                 xlen = strnlen(xname, rem - 1) + 1;
3720                 if (strcmp(XATTR_NAME_LMA, xname) == 0 ||
3721                     strcmp(XATTR_NAME_LMV, xname) == 0)
3722                         continue;
3723
3724                 if (skip_linkea &&
3725                     strcmp(XATTR_NAME_LINK, xname) == 0)
3726                         continue;
3727
3728                 if (skip_dmv &&
3729                     strcmp(XATTR_NAME_DEFAULT_LMV, xname) == 0)
3730                         continue;
3731
3732                 xsize = mdo_xattr_get(env, sobj, &LU_BUF_NULL, xname);
3733                 if (xsize == -ENODATA)
3734                         continue;
3735                 if (xsize < 0)
3736                         GOTO(fini, rc = xsize);
3737
3738                 OBD_ALLOC_PTR(entry);
3739                 if (!entry)
3740                         GOTO(fini, rc = -ENOMEM);
3741
3742                 lu_buf_alloc(&entry->mxe_buf, xsize);
3743                 if (!entry->mxe_buf.lb_buf) {
3744                         OBD_FREE_PTR(entry);
3745                         GOTO(fini, rc = -ENOMEM);
3746                 }
3747
3748                 rc = mdo_xattr_get(env, sobj, &entry->mxe_buf, xname);
3749                 if (rc < 0) {
3750                         lu_buf_free(&entry->mxe_buf);
3751                         OBD_FREE_PTR(entry);
3752                         if (rc == -ENODATA)
3753                                 continue;
3754                         GOTO(fini, rc);
3755                 }
3756
3757                 entry->mxe_name = xname;
3758                 list_add_tail(&entry->mxe_linkage, &xattrs->mx_list);
3759         }
3760
3761         if (needencxattr && !encxattrfound) {
3762                 xlen = strlen(LL_XATTR_NAME_ENCRYPTION_CONTEXT) + 1;
3763                 strncpy(xname, LL_XATTR_NAME_ENCRYPTION_CONTEXT, xlen);
3764                 rem = xlen;
3765                 GOTO(reloop, 0);
3766         }
3767
3768         RETURN(0);
3769 fini:
3770         mdd_xattrs_fini(xattrs);
3771         RETURN(rc);
3772 }
3773
3774 typedef int (*mdd_xattr_cb)(const struct lu_env *env,
3775                             struct mdd_object *obj,
3776                             const struct lu_buf *buf,
3777                             const char *name,
3778                             int fl, struct thandle *handle);
3779
3780 static int mdd_foreach_xattr(const struct lu_env *env,
3781                              struct mdd_object *tobj,
3782                              struct mdd_xattrs *xattrs,
3783                              struct thandle *handle,
3784                              mdd_xattr_cb cb)
3785 {
3786         struct mdd_xattr_entry *entry;
3787         int rc;
3788
3789         list_for_each_entry(entry, &xattrs->mx_list, mxe_linkage) {
3790                 rc = cb(env, tobj, &entry->mxe_buf, entry->mxe_name, 0, handle);
3791                 if (rc)
3792                         return rc;
3793         }
3794
3795         return 0;
3796 }
3797
3798 typedef int (*mdd_linkea_cb)(const struct lu_env *env,
3799                              struct mdd_object *sobj,
3800                              struct mdd_object *tobj,
3801                              const struct lu_name *sname,
3802                              const struct lu_fid *sfid,
3803                              const struct lu_name *lname,
3804                              const struct lu_fid *fid,
3805                              void *opaque,
3806                              struct thandle *handle);
3807
3808 static int mdd_declare_update_link(const struct lu_env *env,
3809                                    struct mdd_object *sobj,
3810                                    struct mdd_object *tobj,
3811                                    const struct lu_name *tname,
3812                                    const struct lu_fid *tpfid,
3813                                    const struct lu_name *lname,
3814                                    const struct lu_fid *fid,
3815                                    void *unused,
3816                                    struct thandle *handle)
3817 {
3818         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3819         struct mdd_object *pobj;
3820         int rc;
3821
3822         /* ignore tobj */
3823         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3824             !strcmp(tname->ln_name, lname->ln_name))
3825                 return 0;
3826
3827         pobj = mdd_object_find(env, mdd, fid);
3828         if (IS_ERR(pobj))
3829                 return PTR_ERR(pobj);
3830
3831
3832         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
3833         if (!rc)
3834                 rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(tobj),
3835                                               mdd_object_type(sobj),
3836                                               lname->ln_name, handle);
3837         mdd_object_put(env, pobj);
3838         if (rc)
3839                 return rc;
3840
3841         rc = mdo_declare_ref_add(env, tobj, handle);
3842         if (rc)
3843                 return rc;
3844
3845         rc = mdo_declare_ref_del(env, sobj, handle);
3846         return rc;
3847 }
3848
3849 static int mdd_update_link(const struct lu_env *env,
3850                            struct mdd_object *sobj,
3851                            struct mdd_object *tobj,
3852                            const struct lu_name *tname,
3853                            const struct lu_fid *tpfid,
3854                            const struct lu_name *lname,
3855                            const struct lu_fid *fid,
3856                            void *unused,
3857                            struct thandle *handle)
3858 {
3859         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3860         struct mdd_object *pobj;
3861         int rc;
3862
3863         ENTRY;
3864
3865         /* ignore tobj */
3866         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3867             !memcmp(tname->ln_name, lname->ln_name, lname->ln_namelen))
3868                 RETURN(0);
3869
3870         CDEBUG(D_INFO, "update "DFID"/"DNAME":"DFID"\n",
3871                PFID(fid), PNAME(lname), PFID(mdd_object_fid(tobj)));
3872
3873         pobj = mdd_object_find(env, mdd, fid);
3874         if (IS_ERR(pobj)) {
3875                 CWARN("%s: cannot find obj "DFID": %ld\n",
3876                       mdd2obd_dev(mdd)->obd_name, PFID(fid), PTR_ERR(pobj));
3877                 RETURN(PTR_ERR(pobj));
3878         }
3879
3880         if (!mdd_object_exists(pobj)) {
3881                 CDEBUG(D_INFO, DFID" doesn't exist\n", PFID(fid));
3882                 mdd_object_put(env, pobj);
3883                 RETURN(-ENOENT);
3884         }
3885
3886         mdd_write_lock(env, pobj, DT_TGT_PARENT);
3887         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
3888         if (!rc)
3889                 rc = __mdd_index_insert_only(env, pobj, mdd_object_fid(tobj),
3890                                              mdd_object_type(sobj),
3891                                              lname->ln_name, handle);
3892         mdd_write_unlock(env, pobj);
3893         mdd_object_put(env, pobj);
3894         if (rc)
3895                 RETURN(rc);
3896
3897         mdd_write_lock(env, tobj, DT_TGT_CHILD);
3898         rc = mdo_ref_add(env, tobj, handle);
3899         mdd_write_unlock(env, tobj);
3900         if (rc)
3901                 RETURN(rc);
3902
3903         mdd_write_lock(env, sobj, DT_SRC_CHILD);
3904         rc = mdo_ref_del(env, sobj, handle);
3905         mdd_write_unlock(env, sobj);
3906
3907         RETURN(rc);
3908 }
3909
3910 static inline int mdd_fld_lookup(const struct lu_env *env,
3911                                  struct mdd_device *mdd,
3912                                  const struct lu_fid *fid,
3913                                  __u32 *mdt_index)
3914 {
3915         struct lu_seq_range *range = &mdd_env_info(env)->mdi_range;
3916         struct seq_server_site *ss;
3917         int rc;
3918
3919         ss = mdd->mdd_md_dev.md_lu_dev.ld_site->ld_seq_site;
3920
3921         range->lsr_flags = LU_SEQ_RANGE_MDT;
3922         rc = fld_server_lookup(env, ss->ss_server_fld, fid->f_seq, range);
3923         if (rc)
3924                 return rc;
3925
3926         *mdt_index = range->lsr_index;
3927
3928         return 0;
3929 }
3930
3931 static int mdd_is_link_on_source_mdt(const struct lu_env *env,
3932                                      struct mdd_object *sobj,
3933                                      struct mdd_object *tobj,
3934                                      const struct lu_name *tname,
3935                                      const struct lu_fid *tpfid,
3936                                      const struct lu_name *lname,
3937                                      const struct lu_fid *fid,
3938                                      void *opaque,
3939                                      struct thandle *handle)
3940 {
3941         struct mdd_device *mdd = mdo2mdd(&sobj->mod_obj);
3942         __u32 source_mdt_index = *(__u32 *)opaque;
3943         __u32 link_mdt_index;
3944         int rc;
3945
3946         ENTRY;
3947
3948         /* ignore tobj */
3949         if (lu_fid_eq(tpfid, fid) && tname->ln_namelen == lname->ln_namelen &&
3950             !memcmp(tname->ln_name, lname->ln_name, lname->ln_namelen))
3951                 return 0;
3952
3953         rc = mdd_fld_lookup(env, mdd, fid, &link_mdt_index);
3954         if (rc)
3955                 RETURN(rc);
3956
3957         RETURN(link_mdt_index == source_mdt_index);
3958 }
3959
3960 static int mdd_iterate_linkea(const struct lu_env *env,
3961                               struct mdd_object *sobj,
3962                               struct mdd_object *tobj,
3963                               const struct lu_name *tname,
3964                               const struct lu_fid *tpfid,
3965                               struct linkea_data *ldata,
3966                               void *opaque,
3967                               struct thandle *handle,
3968                               mdd_linkea_cb cb)
3969 {
3970         struct mdd_thread_info *info = mdd_env_info(env);
3971         char *filename = info->mdi_name;
3972         struct lu_name lname;
3973         struct lu_fid fid;
3974         int rc = 0;
3975
3976         if (!ldata->ld_buf)
3977                 return 0;
3978
3979         for (linkea_first_entry(ldata); ldata->ld_lee && !rc;
3980              linkea_next_entry(ldata)) {
3981                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
3982                                     &fid);
3983
3984                 /* Note: lname might miss \0 at the end */
3985                 snprintf(filename, sizeof(info->mdi_name), "%.*s",
3986                          lname.ln_namelen, lname.ln_name);
3987                 lname.ln_name = filename;
3988
3989                 CDEBUG(D_INFO, DFID"/"DNAME"\n", PFID(&fid), PNAME(&lname));
3990
3991                 rc = cb(env, sobj, tobj, tname, tpfid, &lname, &fid, opaque,
3992                         handle);
3993         }
3994
3995         return rc;
3996 }
3997
3998 /**
3999  * Prepare linkea, and check whether file needs migrate: if source still has
4000  * link on source MDT, no need to migrate, just update namespace on source and
4001  * target parents.
4002  *
4003  * \retval      0 do migrate
4004  * \retval      1 don't migrate
4005  * \retval      -errno on failure
4006  */
4007 static int mdd_migrate_linkea_prepare(const struct lu_env *env,
4008                                       struct mdd_device *mdd,
4009                                       struct mdd_object *spobj,
4010                                       struct mdd_object *tpobj,
4011                                       struct mdd_object *sobj,
4012                                       const struct lu_name *sname,
4013                                       const struct lu_name *tname,
4014                                       const struct lu_attr *attr,
4015                                       struct linkea_data *ldata)
4016 {
4017         __u32 source_mdt_index;
4018         int rc;
4019
4020         ENTRY;
4021
4022         memset(ldata, 0, sizeof(*ldata));
4023         rc = mdd_linkea_prepare(env, sobj, mdd_object_fid(spobj), sname,
4024                                 mdd_object_fid(tpobj), tname, 1, 0, ldata);
4025         if (rc)
4026                 RETURN(rc);
4027
4028         /*
4029          * Then it will check if the file should be migrated. If the file has
4030          * mulitple links, we only need migrate the file if all of its entries
4031          * has been migrated to the remote MDT.
4032          */
4033         if (S_ISDIR(attr->la_mode) || attr->la_nlink < 2)
4034                 RETURN(0);
4035
4036         /* If there are still links locally, don't migrate this file */
4037         LASSERT(ldata->ld_leh != NULL);
4038
4039         /*
4040          * If linkEA is overflow, it means there are some unknown name entries
4041          * under unknown parents, which will prevent the migration.
4042          */
4043         if (unlikely(ldata->ld_leh->leh_overflow_time))
4044                 RETURN(-EOVERFLOW);
4045
4046         rc = mdd_fld_lookup(env, mdd, mdd_object_fid(sobj), &source_mdt_index);
4047         if (rc)
4048                 RETURN(rc);
4049
4050         rc = mdd_iterate_linkea(env, sobj, NULL, tname, mdd_object_fid(tpobj),
4051                                 ldata, &source_mdt_index, NULL,
4052                                 mdd_is_link_on_source_mdt);
4053         RETURN(rc);
4054 }
4055
4056 static int mdd_declare_migrate_update(const struct lu_env *env,
4057                                       struct mdd_object *spobj,
4058                                       struct mdd_object *tpobj,
4059                                       struct mdd_object *obj,
4060                                       const struct lu_name *sname,
4061                                       const struct lu_name *tname,
4062                                       struct lu_attr *attr,
4063                                       struct lu_attr *spattr,
4064                                       struct lu_attr *tpattr,
4065                                       struct linkea_data *ldata,
4066                                       struct md_attr *ma,
4067                                       struct thandle *handle)
4068 {
4069         struct mdd_thread_info *info = mdd_env_info(env);
4070         struct lu_attr *la = &info->mdi_la_for_fix;
4071         int rc;
4072
4073         rc = mdo_declare_index_delete(env, spobj, sname->ln_name, handle);
4074         if (rc)
4075                 return rc;
4076
4077         if (S_ISDIR(attr->la_mode)) {
4078                 rc = mdo_declare_ref_del(env, spobj, handle);
4079                 if (rc)
4080                         return rc;
4081         }
4082
4083         rc = mdo_declare_index_insert(env, tpobj, mdd_object_fid(obj),
4084                                       attr->la_mode & S_IFMT,
4085                                       tname->ln_name, handle);
4086         if (rc)
4087                 return rc;
4088
4089         rc = mdd_declare_links_add(env, obj, handle, ldata);
4090         if (rc)
4091                 return rc;
4092
4093         if (S_ISDIR(attr->la_mode)) {
4094                 rc = mdo_declare_ref_add(env, tpobj, handle);
4095                 if (rc)
4096                         return rc;
4097         }
4098
4099         la->la_valid = LA_CTIME | LA_MTIME;
4100         rc = mdo_declare_attr_set(env, spobj, la, handle);
4101         if (rc)
4102                 return rc;
4103
4104         if (tpobj != spobj) {
4105                 rc = mdo_declare_attr_set(env, tpobj, la, handle);
4106                 if (rc)
4107                         return rc;
4108         }
4109
4110         return rc;
4111 }
4112
4113 static int mdd_declare_migrate_create(const struct lu_env *env,
4114                                       struct mdd_object *spobj,
4115                                       struct mdd_object *tpobj,
4116                                       struct mdd_object *sobj,
4117                                       struct mdd_object *tobj,
4118                                       const struct lu_name *sname,
4119                                       const struct lu_name *tname,
4120                                       struct lu_attr *spattr,
4121                                       struct lu_attr *tpattr,
4122                                       struct lu_attr *attr,
4123                                       struct lu_buf *sbuf,
4124                                       struct linkea_data *ldata,
4125                                       struct mdd_xattrs *xattrs,
4126                                       struct md_attr *ma,
4127                                       struct md_op_spec *spec,
4128                                       struct dt_allocation_hint *hint,
4129                                       struct thandle *handle)
4130 {
4131         struct mdd_thread_info *info = mdd_env_info(env);
4132         struct md_layout_change *mlc = &info->mdi_mlc;
4133         struct lmv_mds_md_v1 *lmv = sbuf->lb_buf;
4134         int rc;
4135
4136         ENTRY;
4137
4138         if (S_ISDIR(attr->la_mode)) {
4139                 struct lmv_user_md *lum = spec->u.sp_ea.eadata;
4140
4141                 mlc->mlc_opc = MD_LAYOUT_DETACH;
4142                 rc = mdo_declare_layout_change(env, sobj, mlc, handle);
4143                 if (rc)
4144                         return rc;
4145
4146                 lum->lum_hash_type |= cpu_to_le32(LMV_HASH_FLAG_MIGRATION);
4147         } else if (S_ISLNK(attr->la_mode)) {
4148                 spec->u.sp_symname.ln_name = sbuf->lb_buf;
4149                 /* don't count NUL */
4150                 spec->u.sp_symname.ln_namelen = sbuf->lb_len - 1;
4151         } else if (S_ISREG(attr->la_mode)) {
4152                 spec->sp_cr_flags |= MDS_OPEN_DELAY_CREATE;
4153                 spec->sp_cr_flags &= ~MDS_OPEN_HAS_EA;
4154         }
4155
4156         mdd_object_make_hint(env, tpobj, tobj, attr, spec, hint);
4157
4158         rc = mdd_declare_create(env, mdo2mdd(&tpobj->mod_obj), tpobj, tobj,
4159                                 tname, attr, handle, spec, ldata, NULL, NULL,
4160                                 NULL, hint);
4161         if (rc)
4162                 return rc;
4163
4164         /*
4165          * tobj mode will be used in mdo_declare_layout_change(), but it's not
4166          * createb yet, copy from sobj.
4167          */
4168         tobj->mod_obj.mo_lu.lo_header->loh_attr &= ~S_IFMT;
4169         tobj->mod_obj.mo_lu.lo_header->loh_attr |=
4170                 sobj->mod_obj.mo_lu.lo_header->loh_attr & S_IFMT;
4171
4172         if (S_ISDIR(attr->la_mode)) {
4173                 if (!lmv) {
4174                         /* if sobj is not striped, fake a 1-stripe LMV */
4175                         LASSERT(sizeof(info->mdi_key) >
4176                                 lmv_mds_md_size(1, LMV_MAGIC_V1));
4177                         lmv = (typeof(lmv))info->mdi_key;
4178                         memset(lmv, 0, sizeof(*lmv));
4179                         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
4180                         lmv->lmv_stripe_count = cpu_to_le32(1);
4181                         lmv->lmv_hash_type = cpu_to_le32(LMV_HASH_TYPE_DEFAULT);
4182                         fid_le_to_cpu(&lmv->lmv_stripe_fids[0],
4183                                       mdd_object_fid(sobj));
4184                         mlc->mlc_buf.lb_buf = lmv;
4185                         mlc->mlc_buf.lb_len = lmv_mds_md_size(1, LMV_MAGIC_V1);
4186                 } else {
4187                         mlc->mlc_buf = *sbuf;
4188                 }
4189                 mlc->mlc_opc = MD_LAYOUT_ATTACH;
4190                 rc = mdo_declare_layout_change(env, tobj, mlc, handle);
4191                 if (rc)
4192                         return rc;
4193         }
4194
4195         rc = mdd_foreach_xattr(env, tobj, xattrs, handle,
4196                                mdo_declare_xattr_set);
4197         if (rc)
4198                 return rc;
4199
4200         if (S_ISREG(attr->la_mode)) {
4201                 struct lu_buf fid_buf;
4202
4203                 handle->th_complex = 1;
4204
4205                 /* target may be remote, update PFID via sobj. */
4206                 fid_buf.lb_buf = (void *)mdd_object_fid(tobj);
4207                 fid_buf.lb_len = sizeof(struct lu_fid);
4208                 rc = mdo_declare_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID,
4209                                            0, handle);
4210                 if (rc)
4211                         return rc;
4212
4213                 rc = mdo_declare_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4214                 if (rc)
4215                         return rc;
4216         }
4217
4218         if (!S_ISDIR(attr->la_mode)) {
4219                 rc = mdd_iterate_linkea(env, sobj, tobj, tname,
4220                                         mdd_object_fid(tpobj), ldata, NULL,
4221                                         handle, mdd_declare_update_link);
4222                 if (rc)
4223                         return rc;
4224         }
4225
4226         if (!S_ISDIR(attr->la_mode) || lmv) {
4227                 rc = mdo_declare_ref_del(env, sobj, handle);
4228                 if (rc)
4229                         return rc;
4230
4231                 if (S_ISDIR(attr->la_mode)) {
4232                         rc = mdo_declare_ref_del(env, sobj, handle);
4233                         if (rc)
4234                                 return rc;
4235                 }
4236
4237                 rc = mdo_declare_destroy(env, sobj, handle);
4238                 if (rc)
4239                         return rc;
4240         }
4241
4242         rc = mdd_declare_migrate_update(env, spobj, tpobj, tobj, sname, tname,
4243                                         attr, spattr, tpattr, ldata, ma,
4244                                         handle);
4245         return rc;
4246 }
4247
4248 /**
4249  * migrate dirent from \a spobj to \a tpobj.
4250  **/
4251 static int mdd_migrate_update(const struct lu_env *env,
4252                               struct mdd_object *spobj,
4253                               struct mdd_object *tpobj,
4254                               struct mdd_object *obj,
4255                               const struct lu_name *sname,
4256                               const struct lu_name *tname,
4257                               struct lu_attr *attr,
4258                               struct lu_attr *spattr,
4259                               struct lu_attr *tpattr,
4260                               struct linkea_data *ldata,
4261                               struct md_attr *ma,
4262                               struct thandle *handle)
4263 {
4264         struct mdd_thread_info *info = mdd_env_info(env);
4265         struct lu_attr *la = &info->mdi_la_for_fix;
4266         int rc;
4267
4268         ENTRY;
4269
4270         CDEBUG(D_INFO, "update "DFID" from "DFID"/%s to "DFID"/%s\n",
4271                PFID(mdd_object_fid(obj)), PFID(mdd_object_fid(spobj)),
4272                sname->ln_name, PFID(mdd_object_fid(tpobj)), tname->ln_name);
4273
4274         rc = __mdd_index_delete(env, spobj, sname->ln_name,
4275                                 S_ISDIR(attr->la_mode), handle);
4276         if (rc)
4277                 RETURN(rc);
4278
4279         rc = __mdd_index_insert(env, tpobj, mdd_object_fid(obj),
4280                                 attr->la_mode & S_IFMT,
4281                                 tname->ln_name, handle);
4282         if (rc)
4283                 RETURN(rc);
4284
4285         rc = mdd_links_write(env, obj, ldata, handle);
4286         if (rc)
4287                 RETURN(rc);
4288
4289         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
4290         la->la_valid = LA_CTIME | LA_MTIME;
4291         mdd_write_lock(env, spobj, DT_SRC_PARENT);
4292         rc = mdd_update_time(env, spobj, spattr, la, handle);
4293         mdd_write_unlock(env, spobj);
4294         if (rc)
4295                 RETURN(rc);
4296
4297         if (tpobj != spobj) {
4298                 la->la_valid = LA_CTIME | LA_MTIME;
4299                 mdd_write_lock(env, tpobj, DT_TGT_PARENT);
4300                 rc = mdd_update_time(env, tpobj, tpattr, la, handle);
4301                 mdd_write_unlock(env, tpobj);
4302                 if (rc)
4303                         RETURN(rc);
4304         }
4305
4306         RETURN(rc);
4307 }
4308
4309 /**
4310  * Migrate file/dir to target MDT.
4311  *
4312  * Create target according to \a spec, and then migrate xattrs, if it's
4313  * directory, migrate source stripes to target.
4314  *
4315  * \param[in] env       execution environment
4316  * \param[in] spobj     source parent object
4317  * \param[in] tpobj     target parent object
4318  * \param[in] sobj      source object
4319  * \param[in] tobj      target object
4320  * \param[in] lname     file name
4321  * \param[in] spattr    source parent attributes
4322  * \param[in] tpattr    target parent attributes
4323  * \param[in] attr      source attributes
4324  * \param[in] sbuf      source LMV buf
4325  * \param[in] spec      migrate create spec
4326  * \param[in] hint      target creation hint
4327  * \param[in] handle    tranasction handle
4328  *
4329  * \retval      0 on success
4330  * \retval      -errno on failure
4331  **/
4332 static int mdd_migrate_create(const struct lu_env *env,
4333                               struct mdd_object *spobj,
4334                               struct mdd_object *tpobj,
4335                               struct mdd_object *sobj,
4336                               struct mdd_object *tobj,
4337                               const struct lu_name *sname,
4338                               const struct lu_name *tname,
4339                               struct lu_attr *spattr,
4340                               struct lu_attr *tpattr,
4341                               struct lu_attr *attr,
4342                               const struct lu_buf *sbuf,
4343                               struct linkea_data *ldata,
4344                               struct mdd_xattrs *xattrs,
4345                               struct md_attr *ma,
4346                               struct md_op_spec *spec,
4347                               struct dt_allocation_hint *hint,
4348                               struct thandle *handle)
4349 {
4350         int rc;
4351
4352         ENTRY;
4353
4354         /*
4355          * migrate sobj stripes to tobj if it's directory:
4356          * 1. detach stripes from sobj.
4357          * 2. attach stripes to tobj, see mdd_declare_migrate_mdt().
4358          * 3. create stripes for tobj, see lod_xattr_set_lmv().
4359          */
4360         if (S_ISDIR(attr->la_mode)) {
4361                 struct mdd_thread_info *info = mdd_env_info(env);
4362                 struct md_layout_change *mlc = &info->mdi_mlc;
4363
4364                 mlc->mlc_opc = MD_LAYOUT_DETACH;
4365
4366                 mdd_write_lock(env, sobj, DT_SRC_PARENT);
4367                 rc = mdo_layout_change(env, sobj, mlc, handle);
4368                 mdd_write_unlock(env, sobj);
4369                 if (rc)
4370                         RETURN(rc);
4371         }
4372
4373         /* don't set nlink from sobj */
4374         attr->la_valid &= ~LA_NLINK;
4375
4376         rc = mdd_create_object(env, tpobj, tobj, attr, spec, NULL, NULL, NULL,
4377                                hint, handle, false);
4378         if (rc)
4379                 RETURN(rc);
4380
4381         mdd_write_lock(env, tobj, DT_TGT_CHILD);
4382         rc = mdd_foreach_xattr(env, tobj, xattrs, handle, mdo_xattr_set);
4383         mdd_write_unlock(env, tobj);
4384         if (rc)
4385                 RETURN(rc);
4386
4387         /* for regular file, update OST objects XATTR_NAME_FID */
4388         if (S_ISREG(attr->la_mode)) {
4389                 struct lu_buf fid_buf;
4390
4391                 /* target may be remote, update PFID via sobj. */
4392                 fid_buf.lb_buf = (void *)mdd_object_fid(tobj);
4393                 fid_buf.lb_len = sizeof(struct lu_fid);
4394                 rc = mdo_xattr_set(env, sobj, &fid_buf, XATTR_NAME_FID, 0,
4395                                    handle);
4396                 if (rc)
4397                         RETURN(rc);
4398
4399                 /* delete LOV to avoid deleting OST objs when destroying sobj */
4400                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4401                 rc = mdo_xattr_del(env, sobj, XATTR_NAME_LOV, handle);
4402                 mdd_write_unlock(env, sobj);
4403                 /* O_DELAY_CREATE file may not have LOV, ignore -ENODATA */
4404                 if (rc && rc != -ENODATA)
4405                         RETURN(rc);
4406                 rc = 0;
4407         }
4408
4409         /* update links FID */
4410         if (!S_ISDIR(attr->la_mode)) {
4411                 rc = mdd_iterate_linkea(env, sobj, tobj, tname,
4412                                         mdd_object_fid(tpobj), ldata,
4413                                         NULL, handle, mdd_update_link);
4414                 if (rc)
4415                         RETURN(rc);
4416         }
4417
4418         /* don't destroy sobj if it's plain directory */
4419         if (!S_ISDIR(attr->la_mode) || sbuf->lb_buf) {
4420                 mdd_write_lock(env, sobj, DT_SRC_CHILD);
4421                 rc = mdo_ref_del(env, sobj, handle);
4422                 if (!rc) {
4423                         if (S_ISDIR(attr->la_mode))
4424                                 rc = mdo_ref_del(env, sobj, handle);
4425                         if (!rc)
4426                                 rc = mdo_destroy(env, sobj, handle);
4427                 }
4428                 mdd_write_unlock(env, sobj);
4429                 if (rc)
4430                         RETURN(rc);
4431         }
4432
4433         rc = mdd_migrate_update(env, spobj, tpobj, tobj, sname, tname, attr,
4434                                 spattr, tpattr, ldata, ma, handle);
4435
4436         RETURN(rc);
4437 }
4438
4439 /* NB: if user issued different migrate command, we can't adjust it silently
4440  * here, because this command will decide target MDT in subdir migration in
4441  * LMV.
4442  */
4443 static int mdd_migrate_cmd_check(struct mdd_device *mdd,
4444                                  const struct lmv_mds_md_v1 *lmv,
4445                                  const struct lmv_user_md_v1 *lum,
4446                                  const struct lu_name *lname)
4447 {
4448         __u32 lum_stripe_count = lum->lum_stripe_count;
4449         __u32 lum_hash_type = lum->lum_hash_type &
4450                               cpu_to_le32(LMV_HASH_TYPE_MASK);
4451         __u32 lmv_hash_type = lmv->lmv_hash_type &
4452                               cpu_to_le32(LMV_HASH_TYPE_MASK);
4453
4454         if (!lmv_is_sane(lmv))
4455                 return -EBADF;
4456
4457         /* if stripe_count unspecified, set to 1 */
4458         if (!lum_stripe_count)
4459                 lum_stripe_count = cpu_to_le32(1);
4460
4461         /* TODO: check specific MDTs */
4462         if (lum_stripe_count != lmv->lmv_migrate_offset ||
4463             lum->lum_stripe_offset != lmv->lmv_master_mdt_index ||
4464             (lum_hash_type && lum_hash_type != lmv_hash_type)) {
4465                 CERROR("%s: '"DNAME"' migration was interrupted, run 'lfs migrate -m %d -c %d -H %s "DNAME"' to finish migration: rc = %d\n",
4466                         mdd2obd_dev(mdd)->obd_name, PNAME(lname),
4467                         le32_to_cpu(lmv->lmv_master_mdt_index),
4468                         le32_to_cpu(lmv->lmv_migrate_offset),
4469                         mdt_hash_name[le32_to_cpu(lmv_hash_type)],
4470                         PNAME(lname), -EPERM);
4471                 return -EPERM;
4472         }
4473
4474         return -EALREADY;
4475 }
4476
4477 /**
4478  * Internal function to migrate directory or file between MDTs.
4479  *
4480  * migrate source to target in following steps:
4481  *   1. create target, append source stripes after target's if it's directory,
4482  *      migrate xattrs and update fid of source links.
4483  *   2. update namespace: migrate dirent from source parent to target parent,
4484  *      update file linkea, and destroy source if it's not needed any more.
4485  *
4486  * \param[in] env       execution environment
4487  * \param[in] spobj     source parent object
4488  * \param[in] tpobj     target parent object
4489  * \param[in] sobj      source object
4490  * \param[in] tobj      target object
4491  * \param[in] sname     source file name
4492  * \param[in] tname     target file name
4493  * \param[in] spec      target creation spec
4494  * \param[in] ma        used to update \a pobj mtime and ctime
4495  *
4496  * \retval              0 on success
4497  * \retval              -errno on failure
4498  */
4499 static int mdd_migrate_object(const struct lu_env *env,
4500                               struct mdd_object *spobj,
4501                               struct mdd_object *tpobj,
4502                               struct mdd_object *sobj,
4503                               struct mdd_object *tobj,
4504                               const struct lu_name *sname,
4505                               const struct lu_name *tname,
4506                               struct md_op_spec *spec,
4507                               struct md_attr *ma)
4508 {
4509         struct mdd_thread_info *info = mdd_env_info(env);
4510         struct mdd_device *mdd = mdo2mdd(&spobj->mod_obj);
4511         struct lu_attr *spattr = &info->mdi_pattr;
4512         struct lu_attr *tpattr = &info->mdi_tpattr;
4513         struct lu_attr *attr = &info->mdi_cattr;
4514         struct linkea_data *ldata = &info->mdi_link_data;
4515         struct dt_allocation_hint *hint = &info->mdi_hint;
4516         struct lu_buf sbuf = { NULL };
4517         struct mdd_xattrs xattrs;
4518         struct lmv_mds_md_v1 *lmv;
4519         struct thandle *handle;
4520         int retried = 0;
4521         int rc;
4522
4523         ENTRY;
4524
4525         CDEBUG(D_INFO, "migrate %s from "DFID"/"DFID" to "DFID"/"DFID"\n",
4526                sname->ln_name, PFID(mdd_object_fid(spobj)),
4527                PFID(mdd_object_fid(sobj)), PFID(mdd_object_fid(tpobj)),
4528                PFID(mdd_object_fid(tobj)));
4529
4530 retry:
4531         rc = mdd_la_get(env, sobj, attr);
4532         if (rc)
4533                 RETURN(rc);
4534
4535         rc = mdd_la_get(env, spobj, spattr);
4536         if (rc)
4537                 RETURN(rc);
4538
4539         rc = mdd_la_get(env, tpobj, tpattr);
4540         if (rc)
4541                 RETURN(rc);
4542
4543         rc = mdd_migrate_sanity_check(env, mdd, spobj, tpobj, sobj, tobj,
4544                                       spattr, tpattr, attr);
4545         if (rc)
4546                 RETURN(rc);
4547
4548         mdd_xattrs_init(&xattrs);
4549
4550         if (S_ISDIR(attr->la_mode) && !spec->sp_migrate_nsonly) {
4551                 struct lmv_user_md_v1 *lum = spec->u.sp_ea.eadata;
4552
4553                 LASSERT(lum);
4554
4555                 /* if user use default value '0' for stripe_count, we need to
4556                  * adjust it to '1' to create a 1-stripe directory.
4557                  */
4558                 if (lum->lum_stripe_count == 0)
4559                         lum->lum_stripe_count = cpu_to_le32(1);
4560
4561                 rc = mdd_stripe_get(env, sobj, &sbuf, XATTR_NAME_LMV);
4562                 if (rc && rc != -ENODATA)
4563                         GOTO(out, rc);
4564
4565                 lmv = sbuf.lb_buf;
4566                 if (lmv) {
4567                         if (!lmv_is_sane(lmv))
4568                                 GOTO(out, rc = -EBADF);
4569                         if (lmv_is_migrating(lmv)) {
4570                                 rc = mdd_migrate_cmd_check(mdd, lmv, lum,
4571                                                            sname);
4572                                 GOTO(out, rc);
4573                         }
4574                 }
4575         } else if (!S_ISDIR(attr->la_mode)) {
4576                 if (spobj == tpobj)
4577                         GOTO(out, rc = -EALREADY);
4578
4579                 /* update namespace only if @sobj is on MDT where @tpobj is. */
4580                 if (!mdd_object_remote(tpobj) && !mdd_object_remote(sobj))
4581                         spec->sp_migrate_nsonly = true;
4582
4583                 if (S_ISLNK(attr->la_mode)) {
4584                         lu_buf_check_and_alloc(&sbuf, attr->la_size + 1);
4585                         if (!sbuf.lb_buf)
4586                                 GOTO(out, rc = -ENOMEM);
4587
4588                         rc = mdd_readlink(env, &sobj->mod_obj, &sbuf);
4589                         if (rc <= 0) {
4590                                 rc = rc ?: -EFAULT;
4591                                 CERROR("%s: "DFID" readlink failed: rc = %d\n",
4592                                        mdd2obd_dev(mdd)->obd_name,
4593                                        PFID(mdd_object_fid(sobj)), rc);
4594                                 GOTO(out, rc);
4595                         }
4596                 }
4597         }
4598
4599         /* linkea needs update upon FID or parent stripe change */
4600         rc = mdd_migrate_linkea_prepare(env, mdd, spobj, tpobj, sobj, sname,
4601                                         tname, attr, ldata);
4602         if (rc > 0)
4603                 /* update namespace only if @sobj has link on its MDT. */
4604                 spec->sp_migrate_nsonly = true;
4605         else if (rc < 0)
4606                 GOTO(out, rc);
4607
4608         /* migrate inode will migrate xattrs, prepare xattrs early to avoid
4609          * RPCs inside transaction.
4610          */
4611         if (!spec->sp_migrate_nsonly) {
4612                 rc = mdd_xattrs_migrate_prep(env, &xattrs, sobj, true, true);
4613                 if (rc)
4614                         GOTO(out, rc);
4615         }
4616
4617         handle = mdd_trans_create(env, mdd);
4618         if (IS_ERR(handle))
4619                 GOTO(out, rc = PTR_ERR(handle));
4620
4621         if (spec->sp_migrate_nsonly)
4622                 rc = mdd_declare_migrate_update(env, spobj, tpobj, sobj, sname,
4623                                                 tname, attr, spattr, tpattr,
4624                                                 ldata, ma, handle);
4625         else
4626                 rc = mdd_declare_migrate_create(env, spobj, tpobj, sobj, tobj,
4627                                                 sname, tname, spattr, tpattr,
4628                                                 attr, &sbuf, ldata, &xattrs, ma,
4629                                                 spec, hint, handle);
4630         if (rc)
4631                 GOTO(stop, rc);
4632
4633         rc = mdd_declare_changelog_store(env, mdd, CL_MIGRATE, tname, sname,
4634                                          handle);
4635         if (rc)
4636                 GOTO(stop, rc);
4637
4638         rc = mdd_trans_start(env, mdd, handle);
4639         if (rc)
4640                 GOTO(stop, rc);
4641
4642         if (spec->sp_migrate_nsonly)
4643                 rc = mdd_migrate_update(env, spobj, tpobj, sobj, sname, tname,
4644                                         attr, spattr, tpattr, ldata, ma,
4645                                         handle);
4646         else
4647                 rc = mdd_migrate_create(env, spobj, tpobj, sobj, tobj, sname,
4648                                         tname, spattr, tpattr, attr, &sbuf,
4649                                         ldata, &xattrs, ma, spec, hint, handle);
4650         if (rc)
4651                 GOTO(stop, rc);
4652
4653         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0,
4654                                     spec->sp_migrate_nsonly ? sobj : tobj,
4655                                     spobj, spattr, mdd_object_fid(sobj),
4656                                     tpobj, tpattr, tname, sname,
4657                                     handle);
4658
4659 stop:
4660         rc = mdd_trans_stop(env, mdd, rc, handle);
4661 out:
4662         mdd_xattrs_fini(&xattrs);
4663         lu_buf_free(&sbuf);
4664
4665         /**
4666          * -EAGAIN means transaction execution phase detect the layout
4667          * has been changed by others.
4668          */
4669         if (rc == -EAGAIN && retried++ < MAX_TRANS_RETRIED)
4670                 GOTO(retry, retried);
4671
4672         RETURN(rc);
4673 }
4674
4675 /**
4676  * Migrate directory or file between MDTs.
4677  *
4678  * \param[in] env       execution environment
4679  * \param[in] md_spobj  source parent object
4680  * \param[in] md_tpobj  target parent object
4681  * \param[in] md_sobj   source object
4682  * \param[in] lname     file name
4683  * \param[in] md_tobj   target object
4684  * \param[in] spec      target creation spec
4685  * \param[in] ma        used to update \a pobj mtime and ctime
4686  *
4687  * \retval              0 on success
4688  * \retval              -errno on failure
4689  */
4690 static int mdd_migrate(const struct lu_env *env, struct md_object *md_spobj,
4691                        struct md_object *md_tpobj, struct md_object *md_sobj,
4692                        struct md_object *md_tobj, const struct lu_name *lname,
4693                        struct md_op_spec *spec, struct md_attr *ma)
4694 {
4695         return mdd_migrate_object(env, md2mdd_obj(md_spobj),
4696                                   md2mdd_obj(md_tpobj), md2mdd_obj(md_sobj),
4697                                   md2mdd_obj(md_tobj), lname, lname, spec, ma);
4698 }
4699
4700 static int mdd_declare_1sd_collapse(const struct lu_env *env,
4701                                     struct mdd_object *pobj,
4702                                     struct mdd_object *obj,
4703                                     struct mdd_object *stripe,
4704                                     struct lu_attr *attr,
4705                                     struct mdd_xattrs *xattrs,
4706                                     struct md_layout_change *mlc,
4707                                     struct lu_name *lname,
4708                                     struct thandle *handle)
4709 {
4710         int rc;
4711
4712         mlc->mlc_opc = MD_LAYOUT_DETACH;
4713         rc = mdo_declare_layout_change(env, obj, mlc, handle);
4714         if (rc)
4715                 return rc;
4716
4717         rc = mdo_declare_index_insert(env, stripe, mdd_object_fid(pobj),
4718                                       S_IFDIR, dotdot, handle);
4719         if (rc)
4720                 return rc;
4721
4722         rc = mdd_foreach_xattr(env, stripe, xattrs, handle,
4723                                mdo_declare_xattr_set);
4724         if (rc)
4725                 return rc;
4726
4727         rc = mdo_declare_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4728         if (rc)
4729                 return rc;
4730
4731         rc = mdo_declare_attr_set(env, stripe, attr, handle);
4732         if (rc)
4733                 return rc;
4734
4735         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
4736         if (rc)
4737                 return rc;
4738
4739         rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(stripe),
4740                                       attr->la_mode, lname->ln_name, handle);
4741         if (rc)
4742                 return rc;
4743
4744         rc = mdo_declare_ref_del(env, obj, handle);
4745         if (rc)
4746                 return rc;
4747
4748         rc = mdo_declare_ref_del(env, obj, handle);
4749         if (rc)
4750                 return rc;
4751
4752         rc = mdo_declare_destroy(env, obj, handle);
4753         if (rc)
4754                 return rc;
4755
4756         return rc;
4757 }
4758
4759 /* transform one-stripe directory to a plain directory */
4760 static int mdd_1sd_collapse(const struct lu_env *env,
4761                             struct mdd_object *pobj,
4762                             struct mdd_object *obj,
4763                             struct mdd_object *stripe,
4764                             struct lu_attr *attr,
4765                             struct mdd_xattrs *xattrs,
4766                             struct md_layout_change *mlc,
4767                             struct lu_name *lname,
4768                             struct thandle *handle)
4769 {
4770         int rc;
4771
4772         ENTRY;
4773
4774         /* replace 1-stripe directory with its stripe */
4775         mlc->mlc_opc = MD_LAYOUT_DETACH;
4776
4777         mdd_write_lock(env, obj, DT_SRC_PARENT);
4778         rc = mdo_layout_change(env, obj, mlc, handle);
4779         mdd_write_unlock(env, obj);
4780         if (rc)
4781                 RETURN(rc);
4782
4783         mdd_write_lock(env, pobj, DT_SRC_PARENT);
4784         mdd_write_lock(env, obj, DT_SRC_CHILD);
4785
4786         /* insert dotdot to stripe which points to parent */
4787         rc = __mdd_index_insert_only(env, stripe, mdd_object_fid(pobj),
4788                                      S_IFDIR, dotdot, handle);
4789         if (rc)
4790                 GOTO(out, rc);
4791
4792         rc = mdd_foreach_xattr(env, stripe, xattrs, handle, mdo_xattr_set);
4793         if (rc)
4794                 GOTO(out, rc);
4795
4796         /* delete LMV */
4797         rc = mdo_xattr_del(env, stripe, XATTR_NAME_LMV, handle);
4798         if (rc)
4799                 GOTO(out, rc);
4800
4801         /* don't set nlink from parent */
4802         attr->la_valid &= ~LA_NLINK;
4803
4804         rc = mdo_attr_set(env, stripe, attr, handle);
4805         if (rc)
4806                 GOTO(out, rc);
4807
4808         /* delete dir name from parent */
4809         rc = __mdd_index_delete_only(env, pobj, lname->ln_name, handle);
4810         if (rc)
4811                 GOTO(out, rc);
4812
4813         /* insert stripe to parent with dir name */
4814         rc = __mdd_index_insert_only(env, pobj, mdd_object_fid(stripe),
4815                                      attr->la_mode, lname->ln_name, handle);
4816         if (rc)
4817                 GOTO(out, rc);
4818
4819         /* destroy dir obj */
4820         rc = mdo_ref_del(env, obj, handle);
4821         if (rc)
4822                 GOTO(out, rc);
4823
4824         rc = mdo_ref_del(env, obj, handle);
4825         if (rc)
4826                 GOTO(out, rc);
4827
4828         rc = mdo_destroy(env, obj, handle);
4829         if (rc)
4830                 GOTO(out, rc);
4831
4832         EXIT;
4833 out:
4834         mdd_write_unlock(env, obj);
4835         mdd_write_unlock(env, pobj);
4836
4837         return rc;
4838 }
4839
4840 /*
4841  * shrink directory stripes after migration/merge
4842  */
4843 int mdd_dir_layout_shrink(const struct lu_env *env,
4844                           struct md_object *md_obj,
4845                           struct md_layout_change *mlc)
4846 {
4847         struct mdd_device *mdd = mdo2mdd(md_obj);
4848         struct mdd_thread_info *info = mdd_env_info(env);
4849         struct mdd_object *obj = md2mdd_obj(md_obj);
4850         struct mdd_object *pobj = NULL;
4851         struct mdd_object *stripe = NULL;
4852         struct lu_attr *attr = &info->mdi_pattr;
4853         struct lu_fid *fid = &info->mdi_fid2;
4854         struct lu_name lname = { NULL };
4855         struct lu_buf lmv_buf = { NULL };
4856         struct mdd_xattrs xattrs;
4857         struct lmv_mds_md_v1 *lmv;
4858         struct lmv_user_md *lmu;
4859         struct thandle *handle;
4860         int rc;
4861
4862         ENTRY;
4863
4864         rc = mdd_la_get(env, obj, attr);
4865         if (rc)
4866                 RETURN(rc);
4867
4868         if (!S_ISDIR(attr->la_mode))
4869                 RETURN(-ENOTDIR);
4870
4871         rc = mdd_stripe_get(env, obj, &lmv_buf, XATTR_NAME_LMV);
4872         if (rc < 0)
4873                 RETURN(rc);
4874
4875         lmv = lmv_buf.lb_buf;
4876         if (!lmv_is_sane(lmv))
4877                 RETURN(-EBADF);
4878
4879         lmu = mlc->mlc_buf.lb_buf;
4880
4881         /* adjust the default value '0' to '1' */
4882         if (lmu->lum_stripe_count == 0)
4883                 lmu->lum_stripe_count = cpu_to_le32(1);
4884
4885         /* these were checked in MDT */
4886         LASSERT(le32_to_cpu(lmu->lum_stripe_count) <
4887                 le32_to_cpu(lmv->lmv_stripe_count));
4888         LASSERT(!lmv_is_splitting(lmv));
4889         LASSERT(lmv_is_migrating(lmv) || lmv_is_merging(lmv));
4890
4891         mdd_xattrs_init(&xattrs);
4892
4893         /* if dir stripe count will be shrunk to 1, it needs to be transformed
4894          * to a plain dir, which will cause FID change and namespace update.
4895          */
4896         if (le32_to_cpu(lmu->lum_stripe_count) == 1) {
4897                 struct linkea_data *ldata = &info->mdi_link_data;
4898                 char *filename = info->mdi_name;
4899
4900                 rc = mdd_links_read(env, obj, ldata);
4901                 if (rc)
4902                         GOTO(out, rc);
4903
4904                 if (ldata->ld_leh->leh_reccount > 1)
4905                         GOTO(out, rc = -EINVAL);
4906
4907                 linkea_first_entry(ldata);
4908                 if (!ldata->ld_lee)
4909                         GOTO(out, rc = -ENODATA);
4910
4911                 linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &lname,
4912                                     fid);
4913
4914                 /* Note: lname might miss \0 at the end */
4915                 snprintf(filename, sizeof(info->mdi_name), "%.*s",
4916                          lname.ln_namelen, lname.ln_name);
4917                 lname.ln_name = filename;
4918
4919                 pobj = mdd_object_find(env, mdd, fid);
4920                 if (IS_ERR(pobj)) {
4921                         rc = PTR_ERR(pobj);
4922                         pobj = NULL;
4923                         GOTO(out, rc);
4924                 }
4925
4926                 fid_le_to_cpu(fid, &lmv->lmv_stripe_fids[0]);
4927
4928                 stripe = mdd_object_find(env, mdd, fid);
4929                 if (IS_ERR(stripe)) {
4930                         mdd_object_put(env, pobj);
4931                         pobj = NULL;
4932                         GOTO(out, rc = PTR_ERR(stripe));
4933                 }
4934
4935                 if (!lmv_is_fixed(lmv))
4936                         rc = mdd_xattrs_migrate_prep(env, &xattrs, obj, false,
4937                                                      false);
4938         }
4939
4940         handle = mdd_trans_create(env, mdd);
4941         if (IS_ERR(handle))
4942                 GOTO(out, rc = PTR_ERR(handle));
4943
4944         mlc->mlc_opc = MD_LAYOUT_SHRINK;
4945         rc = mdo_declare_layout_change(env, obj, mlc, handle);
4946         if (rc)
4947                 GOTO(stop_trans, rc);
4948
4949         if (le32_to_cpu(lmu->lum_stripe_count) == 1 && !lmv_is_fixed(lmv)) {
4950                 rc = mdd_declare_1sd_collapse(env, pobj, obj, stripe, attr,
4951                                               &xattrs, mlc, &lname, handle);
4952                 if (rc)
4953                         GOTO(stop_trans, rc);
4954         }
4955
4956         rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL, NULL,
4957                                          handle);
4958         if (rc)
4959                 GOTO(stop_trans, rc);
4960
4961         rc = mdd_trans_start(env, mdd, handle);
4962         if (rc)
4963                 GOTO(stop_trans, rc);
4964
4965         mdd_write_lock(env, obj, DT_SRC_PARENT);
4966         mlc->mlc_opc = MD_LAYOUT_SHRINK;
4967         rc = mdo_layout_change(env, obj, mlc, handle);
4968         mdd_write_unlock(env, obj);
4969         if (rc)
4970                 GOTO(stop_trans, rc);
4971
4972         if (le32_to_cpu(lmu->lum_stripe_count) == 1 && !lmv_is_fixed(lmv)) {
4973                 rc = mdd_1sd_collapse(env, pobj, obj, stripe, attr, &xattrs,
4974                                       mlc, &lname, handle);
4975                 if (rc)
4976                         GOTO(stop_trans, rc);
4977         }
4978
4979         rc = mdd_changelog_data_store_xattr(env, mdd, CL_LAYOUT, 0, obj,
4980                                             XATTR_NAME_LMV, handle);
4981         GOTO(stop_trans, rc);
4982
4983 stop_trans:
4984         rc = mdd_trans_stop(env, mdd, rc, handle);
4985 out:
4986         mdd_xattrs_fini(&xattrs);
4987         if (pobj) {
4988                 mdd_object_put(env, stripe);
4989                 mdd_object_put(env, pobj);
4990         }
4991         lu_buf_free(&lmv_buf);
4992         return rc;
4993 }
4994
4995 static int mdd_dir_declare_split_plain(const struct lu_env *env,
4996                                         struct mdd_device *mdd,
4997                                         struct mdd_object *pobj,
4998                                         struct mdd_object *obj,
4999                                         struct mdd_object *tobj,
5000                                         struct mdd_xattrs *xattrs,
5001                                         struct md_layout_change *mlc,
5002                                         struct dt_allocation_hint *hint,
5003                                         struct thandle *handle)
5004 {
5005         struct mdd_thread_info *info = mdd_env_info(env);
5006         const struct lu_name *lname = mlc->mlc_name;
5007         struct lu_attr *la = &info->mdi_la_for_fix;
5008         struct lmv_user_md_v1 *lum = mlc->mlc_spec->u.sp_ea.eadata;
5009         struct linkea_data *ldata = &info->mdi_link_data;
5010         struct lmv_mds_md_v1 *lmv;
5011         __u32 count;
5012         int rc;
5013
5014         mlc->mlc_opc = MD_LAYOUT_DETACH;
5015         rc = mdo_declare_layout_change(env, obj, mlc, handle);
5016         if (rc)
5017                 return rc;
5018
5019         memset(ldata, 0, sizeof(*ldata));
5020         rc = mdd_linkea_prepare(env, obj, NULL, NULL, mdd_object_fid(pobj),
5021                                 lname, 1, 0, ldata);
5022         if (rc)
5023                 return rc;
5024
5025         count = lum->lum_stripe_count;
5026         lum->lum_stripe_count = 0;
5027         /* don't set default LMV since it will become a striped dir  */
5028         lum->lum_max_inherit = LMV_INHERIT_NONE;
5029         mdd_object_make_hint(env, pobj, tobj, mlc->mlc_attr, mlc->mlc_spec,
5030                              hint);
5031         rc = mdd_declare_create(env, mdo2mdd(&pobj->mod_obj), pobj, tobj,
5032                                 lname, mlc->mlc_attr, handle, mlc->mlc_spec,
5033                                 ldata, NULL, NULL, NULL, hint);
5034         if (rc)
5035                 return rc;
5036
5037         /* tobj mode will be used in lod_declare_xattr_set(), but it's not
5038          * created yet.
5039          */
5040         tobj->mod_obj.mo_lu.lo_header->loh_attr |= S_IFDIR;
5041
5042         lmv = (typeof(lmv))info->mdi_key;
5043         memset(lmv, 0, sizeof(*lmv));
5044         lmv->lmv_magic = cpu_to_le32(LMV_MAGIC_V1);
5045         lmv->lmv_stripe_count = cpu_to_le32(1);
5046         lmv->lmv_hash_type = cpu_to_le32(LMV_HASH_TYPE_DEFAULT);
5047         fid_le_to_cpu(&lmv->lmv_stripe_fids[0], mdd_object_fid(obj));
5048
5049         mlc->mlc_opc = MD_LAYOUT_ATTACH;
5050         mlc->mlc_buf.lb_buf = lmv;
5051         mlc->mlc_buf.lb_len = lmv_mds_md_size(1, LMV_MAGIC_V1);
5052         rc = mdo_declare_layout_change(env, tobj, mlc, handle);
5053         if (rc)
5054                 return rc;
5055
5056         rc = mdd_foreach_xattr(env, tobj, xattrs, handle,
5057                                mdo_declare_xattr_set);
5058         if (rc)
5059                 return rc;
5060
5061         lum->lum_stripe_count = count;
5062         mlc->mlc_opc = MD_LAYOUT_SPLIT;
5063         rc = mdo_declare_layout_change(env, tobj, mlc, handle);
5064         if (rc)
5065                 return rc;
5066
5067         rc = mdo_declare_index_delete(env, pobj, lname->ln_name, handle);
5068         if (rc)
5069                 return rc;
5070
5071         rc = mdo_declare_index_insert(env, pobj, mdd_object_fid(tobj),
5072                                       S_IFDIR, lname->ln_name, handle);
5073         if (rc)
5074                 return rc;
5075
5076         la->la_valid = LA_CTIME | LA_MTIME;
5077         rc = mdo_declare_attr_set(env, obj, la, handle);
5078         if (rc)
5079                 return rc;
5080
5081         rc = mdo_declare_attr_set(env, pobj, la, handle);
5082         if (rc)
5083                 return rc;
5084
5085         rc = mdd_declare_changelog_store(env, mdd, CL_MIGRATE, lname, NULL,
5086                                          handle);
5087         return rc;
5088 }
5089
5090 /**
5091  * plain directory split:
5092  * 1. create \a tobj as plain directory.
5093  * 2. append \a obj as first stripe of \a tobj.
5094  * 3. migrate xattrs from \a obj to \a tobj.
5095  * 4. split \a tobj to specific stripe count.
5096  */
5097 static int mdd_dir_split_plain(const struct lu_env *env,
5098                                 struct mdd_device *mdd,
5099                                 struct mdd_object *pobj,
5100                                 struct mdd_object *obj,
5101                                 struct mdd_object *tobj,
5102                                 struct mdd_xattrs *xattrs,
5103                                 struct md_layout_change *mlc,
5104                                 struct dt_allocation_hint *hint,
5105                                 struct thandle *handle)
5106 {
5107         struct mdd_thread_info *info = mdd_env_info(env);
5108         struct lu_attr *pattr = &info->mdi_pattr;
5109         struct lu_attr *la = &info->mdi_la_for_fix;
5110         const struct lu_name *lname = mlc->mlc_name;
5111         struct linkea_data *ldata = &info->mdi_link_data;
5112         int rc;
5113
5114         ENTRY;
5115
5116         /* copy linkea out and set on target later */
5117         rc = mdd_links_read(env, obj, ldata);
5118         if (rc)
5119                 RETURN(rc);
5120
5121         mlc->mlc_opc = MD_LAYOUT_DETACH;
5122         rc = mdo_layout_change(env, obj, mlc, handle);
5123         if (rc)
5124                 RETURN(rc);
5125
5126         /* don't set nlink from obj */
5127         mlc->mlc_attr->la_valid &= ~LA_NLINK;
5128
5129         rc = mdd_create_object(env, pobj, tobj, mlc->mlc_attr, mlc->mlc_spec,
5130                                NULL, NULL, NULL, hint, handle, false);
5131         if (rc)
5132                 RETURN(rc);
5133
5134         rc = mdd_foreach_xattr(env, tobj, xattrs, handle, mdo_xattr_set);
5135         if (rc)
5136                 RETURN(rc);
5137
5138         rc = mdd_links_write(env, tobj, ldata, handle);
5139         if (rc)
5140                 RETURN(rc);
5141
5142         rc = __mdd_index_delete(env, pobj, lname->ln_name, true, handle);
5143         if (rc)
5144                 RETURN(rc);
5145
5146         rc = __mdd_index_insert(env, pobj, mdd_object_fid(tobj), S_IFDIR,
5147                                 lname->ln_name, handle);
5148         if (rc)
5149                 RETURN(rc);
5150
5151         la->la_ctime = la->la_mtime = mlc->mlc_attr->la_mtime;
5152         la->la_valid = LA_CTIME | LA_MTIME;
5153
5154         mdd_write_lock(env, obj, DT_SRC_CHILD);
5155         rc = mdd_update_time(env, tobj, mlc->mlc_attr, la, handle);
5156         mdd_write_unlock(env, obj);
5157         if (rc)
5158                 RETURN(rc);
5159
5160         rc = mdd_la_get(env, pobj, pattr);
5161         if (rc)
5162                 RETURN(rc);
5163
5164         la->la_valid = LA_CTIME | LA_MTIME;
5165
5166         mdd_write_lock(env, pobj, DT_SRC_PARENT);
5167         rc = mdd_update_time(env, pobj, pattr, la, handle);
5168         mdd_write_unlock(env, pobj);
5169         if (rc)
5170                 RETURN(rc);
5171
5172         /* FID changes, record it as CL_MIGRATE */
5173         rc = mdd_changelog_ns_store(env, mdd, CL_MIGRATE, 0, tobj,
5174                                     pobj, pattr, mdd_object_fid(obj),
5175                                     pobj, pattr, lname, lname, handle);
5176         RETURN(rc);
5177 }
5178
5179 int mdd_dir_layout_split(const struct lu_env *env, struct md_object *o,
5180                          struct md_layout_change *mlc)
5181 {
5182         struct mdd_thread_info *info = mdd_env_info(env);
5183         struct mdd_device *mdd = mdo2mdd(o);
5184         struct mdd_object *obj = md2mdd_obj(o);
5185         struct mdd_object *pobj = md2mdd_obj(mlc->mlc_parent);
5186         struct mdd_object *tobj = md2mdd_obj(mlc->mlc_target);
5187         struct dt_allocation_hint *hint = &info->mdi_hint;
5188         bool is_plain = false;
5189         struct mdd_xattrs xattrs;
5190         struct thandle *handle;
5191         int rc;
5192
5193         ENTRY;
5194
5195         LASSERT(S_ISDIR(mdd_object_type(obj)));
5196
5197         rc = mdo_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LMV);
5198         if (rc == -ENODATA)
5199                 is_plain = true;
5200         else if (rc < 0)
5201                 RETURN(rc);
5202
5203         mdd_xattrs_init(&xattrs);
5204         if (is_plain)
5205                 rc = mdd_xattrs_migrate_prep(env, &xattrs, obj, true, true);
5206
5207         handle = mdd_trans_create(env, mdd);
5208         if (IS_ERR(handle))
5209                 GOTO(out, rc = PTR_ERR(handle));
5210
5211         if (is_plain) {
5212                 rc = mdd_dir_declare_split_plain(env, mdd, pobj, obj, tobj,
5213                                                  &xattrs, mlc, hint, handle);
5214         } else {
5215                 mlc->mlc_opc = MD_LAYOUT_SPLIT;
5216                 rc = mdo_declare_layout_change(env, obj, mlc, handle);
5217                 if (rc)
5218                         GOTO(stop_trans, rc);
5219
5220                 rc = mdd_declare_changelog_store(env, mdd, CL_LAYOUT, NULL,
5221                                                  NULL, handle);
5222         }
5223         if (rc)
5224                 GOTO(stop_trans, rc);
5225
5226         rc = mdd_trans_start(env, mdd, handle);
5227         if (rc)
5228                 GOTO(stop_trans, rc);
5229
5230         if (is_plain) {
5231                 rc = mdd_dir_split_plain(env, mdd, pobj, obj, tobj, &xattrs,
5232                                          mlc, hint, handle);
5233         } else {
5234                 struct lu_buf *buf = &info->mdi_buf[0];
5235
5236                 buf->lb_buf = mlc->mlc_spec->u.sp_ea.eadata;
5237                 buf->lb_len = mlc->mlc_spec->u.sp_ea.eadatalen;
5238
5239                 mdd_write_lock(env, obj, DT_TGT_CHILD);
5240                 rc = mdo_xattr_set(env, obj, buf, XATTR_NAME_LMV,
5241                                    LU_XATTR_CREATE, handle);
5242                 mdd_write_unlock(env, obj);
5243                 if (rc)
5244                         GOTO(stop_trans, rc);
5245
5246                 rc = mdd_changelog_data_store_xattr(env, mdd, CL_LAYOUT, 0, obj,
5247                                                     XATTR_NAME_LMV, handle);
5248         }
5249         if (rc)
5250                 GOTO(stop_trans, rc);
5251
5252         EXIT;
5253
5254 stop_trans:
5255         rc = mdd_trans_stop(env, mdd, rc, handle);
5256 out:
5257         mdd_xattrs_fini(&xattrs);
5258
5259         return rc;
5260 }
5261
5262 const struct md_dir_operations mdd_dir_ops = {
5263         .mdo_is_subdir     = mdd_is_subdir,
5264         .mdo_lookup        = mdd_lookup,
5265         .mdo_create        = mdd_create,
5266         .mdo_rename        = mdd_rename,
5267         .mdo_link          = mdd_link,
5268         .mdo_unlink        = mdd_unlink,
5269         .mdo_create_data   = mdd_create_data,
5270         .mdo_migrate       = mdd_migrate,
5271 };