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