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