lustre/mdd/mdd_dir.c

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