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