lustre/mdd/mdd_dir.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011 Whamcloud, Inc.
  33  *
  34  */
  35 /*
  36  * This file is part of Lustre, http://www.lustre.org/
  37  * Lustre is a trademark of Sun Microsystems, Inc.
  38  *
  39  * lustre/mdd/mdd_dir.c
  40  *
  41  * Lustre Metadata Server (mdd) routines
  42  *
  43  * Author: Wang Di <wangdi@clusterfs.com>
  44  */
  45
  46 #ifndef EXPORT_SYMTAB
  47 # define EXPORT_SYMTAB
  48 #endif
  49 #define DEBUG_SUBSYSTEM S_MDS
  50
  51 #include <linux/module.h>
  52 #ifdef HAVE_EXT4_LDISKFS
  53 #include <ldiskfs/ldiskfs_jbd2.h>
  54 #else
  55 #include <linux/jbd.h>
  56 #endif
  57 #include <obd.h>
  58 #include <obd_class.h>
  59 #include <lustre_ver.h>
  60 #include <obd_support.h>
  61 #include <lprocfs_status.h>
  62 #ifdef HAVE_EXT4_LDISKFS
  63 #include <ldiskfs/ldiskfs.h>
  64 #else
  65 #include <linux/ldiskfs_fs.h>
  66 #endif
  67 #include <lustre_mds.h>
  68 #include <lustre/lustre_idl.h>
  69 #include <lustre_fid.h>
  70
  71 #include "mdd_internal.h"
  72
  73 static const char dot[] = ".";
  74 static const char dotdot[] = "..";
  75
  76 static struct lu_name lname_dotdot = {
  77         (char *) dotdot,
  78         sizeof(dotdot) - 1
  79 };
  80
  81 static int __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
  82                         const struct lu_name *lname, struct lu_fid* fid,
  83                         int mask);
  84 static int mdd_declare_links_add(const struct lu_env *env,
  85                                  struct mdd_object *mdd_obj,
  86                                  struct thandle *handle);
  87 static int mdd_links_add(const struct lu_env *env,
  88                          struct mdd_object *mdd_obj,
  89                          const struct lu_fid *pfid,
  90                          const struct lu_name *lname,
  91                          struct thandle *handle, int first);
  92 static int mdd_links_rename(const struct lu_env *env,
  93                             struct mdd_object *mdd_obj,
  94                             const struct lu_fid *oldpfid,
  95                             const struct lu_name *oldlname,
  96                             const struct lu_fid *newpfid,
  97                             const struct lu_name *newlname,
  98                             struct thandle *handle);
  99
 100 static int
 101 __mdd_lookup_locked(const struct lu_env *env, struct md_object *pobj,
 102                     const struct lu_name *lname, struct lu_fid* fid, int mask)
 103 {
 104         const char *name = lname->ln_name;
 105         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
 106         struct dynlock_handle *dlh;
 107         int rc;
 108
 109         dlh = mdd_pdo_read_lock(env, mdd_obj, name, MOR_TGT_PARENT);
 110         if (unlikely(dlh == NULL))
 111                 return -ENOMEM;
 112         rc = __mdd_lookup(env, pobj, lname, fid, mask);
 113         mdd_pdo_read_unlock(env, mdd_obj, dlh);
 114
 115         return rc;
 116 }
 117
 118 int mdd_lookup(const struct lu_env *env,
 119                struct md_object *pobj, const struct lu_name *lname,
 120                struct lu_fid* fid, struct md_op_spec *spec)
 121 {
 122         int rc;
 123         ENTRY;
 124         rc = __mdd_lookup_locked(env, pobj, lname, fid, MAY_EXEC);
 125         RETURN(rc);
 126 }
 127
 128 static int mdd_parent_fid(const struct lu_env *env, struct mdd_object *obj,
 129                           struct lu_fid *fid)
 130 {
 131         return __mdd_lookup_locked(env, &obj->mod_obj, &lname_dotdot, fid, 0);
 132 }
 133
 134 /*
 135  * For root fid use special function, which does not compare version component
 136  * of fid. Version component is different for root fids on all MDTs.
 137  */
 138 int mdd_is_root(struct mdd_device *mdd, const struct lu_fid *fid)
 139 {
 140         return fid_seq(&mdd->mdd_root_fid) == fid_seq(fid) &&
 141                 fid_oid(&mdd->mdd_root_fid) == fid_oid(fid);
 142 }
 143
 144 /*
 145  * return 1: if lf is the fid of the ancestor of p1;
 146  * return 0: if not;
 147  *
 148  * return -EREMOTE: if remote object is found, in this
 149  * case fid of remote object is saved to @pf;
 150  *
 151  * otherwise: values < 0, errors.
 152  */
 153 static int mdd_is_parent(const struct lu_env *env,
 154                          struct mdd_device *mdd,
 155                          struct mdd_object *p1,
 156                          const struct lu_fid *lf,
 157                          struct lu_fid *pf)
 158 {
 159         struct mdd_object *parent = NULL;
 160         struct lu_fid *pfid;
 161         int rc;
 162         ENTRY;
 163
 164         LASSERT(!lu_fid_eq(mdo2fid(p1), lf));
 165         pfid = &mdd_env_info(env)->mti_fid;
 166
 167         /* Check for root first. */
 168         if (mdd_is_root(mdd, mdo2fid(p1)))
 169                 RETURN(0);
 170
 171         for(;;) {
 172                 /* this is done recursively, bypass capa for each obj */
 173                 mdd_set_capainfo(env, 4, p1, BYPASS_CAPA);
 174                 rc = mdd_parent_fid(env, p1, pfid);
 175                 if (rc)
 176                         GOTO(out, rc);
 177                 if (mdd_is_root(mdd, pfid))
 178                         GOTO(out, rc = 0);
 179                 if (lu_fid_eq(pfid, lf))
 180                         GOTO(out, rc = 1);
 181                 if (parent)
 182                         mdd_object_put(env, parent);
 183                 parent = mdd_object_find(env, mdd, pfid);
 184
 185                 /* cross-ref parent */
 186                 if (parent == NULL) {
 187                         if (pf != NULL)
 188                                 *pf = *pfid;
 189                         GOTO(out, rc = -EREMOTE);
 190                 } else if (IS_ERR(parent))
 191                         GOTO(out, rc = PTR_ERR(parent));
 192                 p1 = parent;
 193         }
 194         EXIT;
 195 out:
 196         if (parent && !IS_ERR(parent))
 197                 mdd_object_put(env, parent);
 198         return rc;
 199 }
 200
 201 /*
 202  * No permission check is needed.
 203  *
 204  * returns 1: if fid is ancestor of @mo;
 205  * returns 0: if fid is not a ancestor of @mo;
 206  *
 207  * returns EREMOTE if remote object is found, fid of remote object is saved to
 208  * @fid;
 209  *
 210  * returns < 0: if error
 211  */
 212 int mdd_is_subdir(const struct lu_env *env, struct md_object *mo,
 213                   const struct lu_fid *fid, struct lu_fid *sfid)
 214 {
 215         struct mdd_device *mdd = mdo2mdd(mo);
 216         int rc;
 217         ENTRY;
 218
 219         if (!S_ISDIR(mdd_object_type(md2mdd_obj(mo))))
 220                 RETURN(0);
 221
 222         rc = mdd_is_parent(env, mdd, md2mdd_obj(mo), fid, sfid);
 223         if (rc == 0) {
 224                 /* found root */
 225                 fid_zero(sfid);
 226         } else if (rc == 1) {
 227                 /* found @fid is parent */
 228                 *sfid = *fid;
 229                 rc = 0;
 230         }
 231         RETURN(rc);
 232 }
 233
 234 /*
 235  * Check that @dir contains no entries except (possibly) dot and dotdot.
 236  *
 237  * Returns:
 238  *
 239  *             0        empty
 240  *      -ENOTDIR        not a directory object
 241  *    -ENOTEMPTY        not empty
 242  *           -ve        other error
 243  *
 244  */
 245 static int mdd_dir_is_empty(const struct lu_env *env,
 246                             struct mdd_object *dir)
 247 {
 248         struct dt_it     *it;
 249         struct dt_object *obj;
 250         const struct dt_it_ops *iops;
 251         int result;
 252         ENTRY;
 253
 254         obj = mdd_object_child(dir);
 255         if (!dt_try_as_dir(env, obj))
 256                 RETURN(-ENOTDIR);
 257
 258         iops = &obj->do_index_ops->dio_it;
 259         it = iops->init(env, obj, LUDA_64BITHASH, BYPASS_CAPA);
 260         if (!IS_ERR(it)) {
 261                 result = iops->get(env, it, (const void *)"");
 262                 if (result > 0) {
 263                         int i;
 264                         for (result = 0, i = 0; result == 0 && i < 3; ++i)
 265                                 result = iops->next(env, it);
 266                         if (result == 0)
 267                                 result = -ENOTEMPTY;
 268                         else if (result == +1)
 269                                 result = 0;
 270                 } else if (result == 0)
 271                         /*
 272                          * Huh? Index contains no zero key?
 273                          */
 274                         result = -EIO;
 275
 276                 iops->put(env, it);
 277                 iops->fini(env, it);
 278         } else
 279                 result = PTR_ERR(it);
 280         RETURN(result);
 281 }
 282
 283 static int __mdd_may_link(const struct lu_env *env, struct mdd_object *obj)
 284 {
 285         struct mdd_device *m = mdd_obj2mdd_dev(obj);
 286         struct lu_attr *la = &mdd_env_info(env)->mti_la;
 287         int rc;
 288         ENTRY;
 289
 290         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
 291         if (rc)
 292                 RETURN(rc);
 293
 294         /*
 295          * Subdir count limitation can be broken through.
 296          */
 297         if (la->la_nlink >= m->mdd_dt_conf.ddp_max_nlink &&
 298             !S_ISDIR(la->la_mode))
 299                 RETURN(-EMLINK);
 300         else
 301                 RETURN(0);
 302 }
 303
 304 /*
 305  * Check whether it may create the cobj under the pobj.
 306  * cobj maybe NULL
 307  */
 308 int mdd_may_create(const struct lu_env *env, struct mdd_object *pobj,
 309                    struct mdd_object *cobj, int check_perm, int check_nlink)
 310 {
 311         int rc = 0;
 312         ENTRY;
 313
 314         if (cobj && mdd_object_exists(cobj))
 315                 RETURN(-EEXIST);
 316
 317         if (mdd_is_dead_obj(pobj))
 318                 RETURN(-ENOENT);
 319
 320         if (check_perm)
 321                 rc = mdd_permission_internal_locked(env, pobj, NULL,
 322                                                     MAY_WRITE | MAY_EXEC,
 323                                                     MOR_TGT_PARENT);
 324
 325         if (!rc && check_nlink)
 326                 rc = __mdd_may_link(env, pobj);
 327
 328         RETURN(rc);
 329 }
 330
 331 /*
 332  * Check whether can unlink from the pobj in the case of "cobj == NULL".
 333  */
 334 int mdd_may_unlink(const struct lu_env *env, struct mdd_object *pobj,
 335                    const struct md_attr *ma)
 336 {
 337         int rc;
 338         ENTRY;
 339
 340         if (mdd_is_dead_obj(pobj))
 341                 RETURN(-ENOENT);
 342
 343         if ((ma->ma_attr.la_valid & LA_FLAGS) &&
 344             (ma->ma_attr.la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL)))
 345                 RETURN(-EPERM);
 346
 347         rc = mdd_permission_internal_locked(env, pobj, NULL,
 348                                             MAY_WRITE | MAY_EXEC,
 349                                             MOR_TGT_PARENT);
 350         if (rc)
 351                 RETURN(rc);
 352
 353         if (mdd_is_append(pobj))
 354                 RETURN(-EPERM);
 355
 356         RETURN(rc);
 357 }
 358
 359 /*
 360  * pobj == NULL is remote ops case, under such case, pobj's
 361  * VTX feature has been checked already, no need check again.
 362  */
 363 static inline int mdd_is_sticky(const struct lu_env *env,
 364                                 struct mdd_object *pobj,
 365                                 struct mdd_object *cobj)
 366 {
 367         struct lu_attr *tmp_la = &mdd_env_info(env)->mti_la;
 368         struct md_ucred *uc = md_ucred(env);
 369         int rc;
 370
 371         if (pobj) {
 372                 rc = mdd_la_get(env, pobj, tmp_la, BYPASS_CAPA);
 373                 if (rc)
 374                         return rc;
 375
 376                 if (!(tmp_la->la_mode & S_ISVTX) ||
 377                      (tmp_la->la_uid == uc->mu_fsuid))
 378                         return 0;
 379         }
 380
 381         rc = mdd_la_get(env, cobj, tmp_la, BYPASS_CAPA);
 382         if (rc)
 383                 return rc;
 384
 385         if (tmp_la->la_uid == uc->mu_fsuid)
 386                 return 0;
 387
 388         return !mdd_capable(uc, CFS_CAP_FOWNER);
 389 }
 390
 391 /*
 392  * Check whether it may delete the cobj from the pobj.
 393  * pobj maybe NULL
 394  */
 395 int mdd_may_delete(const struct lu_env *env, struct mdd_object *pobj,
 396                    struct mdd_object *cobj, struct md_attr *ma,
 397                    int check_perm, int check_empty)
 398 {
 399         int rc = 0;
 400         ENTRY;
 401
 402         LASSERT(cobj);
 403         if (!mdd_object_exists(cobj))
 404                 RETURN(-ENOENT);
 405
 406         if (mdd_is_dead_obj(cobj))
 407                 RETURN(-ESTALE);
 408
 409         if (pobj) {
 410                 if (!mdd_object_exists(pobj))
 411                         RETURN(-ENOENT);
 412
 413                 if (mdd_is_dead_obj(pobj))
 414                         RETURN(-ENOENT);
 415
 416                 if (check_perm) {
 417                         rc = mdd_permission_internal_locked(env, pobj, NULL,
 418                                                     MAY_WRITE | MAY_EXEC,
 419                                                     MOR_TGT_PARENT);
 420                         if (rc)
 421                                 RETURN(rc);
 422                 }
 423
 424                 if (mdd_is_append(pobj))
 425                         RETURN(-EPERM);
 426         }
 427
 428         if (!(ma->ma_attr_flags & MDS_VTX_BYPASS) &&
 429             mdd_is_sticky(env, pobj, cobj))
 430                 RETURN(-EPERM);
 431
 432         if (mdd_is_immutable(cobj) || mdd_is_append(cobj))
 433                 RETURN(-EPERM);
 434
 435         if ((ma->ma_attr.la_valid & LA_FLAGS) &&
 436             (ma->ma_attr.la_flags & (LUSTRE_APPEND_FL | LUSTRE_IMMUTABLE_FL)))
 437                 RETURN(-EPERM);
 438
 439         if (S_ISDIR(ma->ma_attr.la_mode)) {
 440                 struct mdd_device *mdd = mdo2mdd(&cobj->mod_obj);
 441
 442                 if (!S_ISDIR(mdd_object_type(cobj)))
 443                         RETURN(-ENOTDIR);
 444
 445                 if (lu_fid_eq(mdo2fid(cobj), &mdd->mdd_root_fid))
 446                         RETURN(-EBUSY);
 447         } else if (S_ISDIR(mdd_object_type(cobj)))
 448                 RETURN(-EISDIR);
 449
 450         if (S_ISDIR(ma->ma_attr.la_mode) && check_empty)
 451                 rc = mdd_dir_is_empty(env, cobj);
 452
 453         RETURN(rc);
 454 }
 455
 456 /*
 457  * tgt maybe NULL
 458  * has mdd_write_lock on src already, but not on tgt yet
 459  */
 460 int mdd_link_sanity_check(const struct lu_env *env,
 461                           struct mdd_object *tgt_obj,
 462                           const struct lu_name *lname,
 463                           struct mdd_object *src_obj)
 464 {
 465         struct mdd_device *m = mdd_obj2mdd_dev(src_obj);
 466         int rc = 0;
 467         ENTRY;
 468
 469         if (!mdd_object_exists(src_obj))
 470                 RETURN(-ENOENT);
 471
 472         if (mdd_is_dead_obj(src_obj))
 473                 RETURN(-ESTALE);
 474
 475         /* Local ops, no lookup before link, check filename length here. */
 476         if (lname && (lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
 477                 RETURN(-ENAMETOOLONG);
 478
 479         if (mdd_is_immutable(src_obj) || mdd_is_append(src_obj))
 480                 RETURN(-EPERM);
 481
 482         if (S_ISDIR(mdd_object_type(src_obj)))
 483                 RETURN(-EPERM);
 484
 485         LASSERT(src_obj != tgt_obj);
 486         if (tgt_obj) {
 487                 rc = mdd_may_create(env, tgt_obj, NULL, 1, 0);
 488                 if (rc)
 489                         RETURN(rc);
 490         }
 491
 492         rc = __mdd_may_link(env, src_obj);
 493
 494         RETURN(rc);
 495 }
 496
 497 static int __mdd_index_delete_only(const struct lu_env *env, struct mdd_object *pobj,
 498                                    const char *name, struct thandle *handle,
 499                                    struct lustre_capa *capa)
 500 {
 501         struct dt_object *next = mdd_object_child(pobj);
 502         int               rc;
 503         ENTRY;
 504
 505         if (dt_try_as_dir(env, next)) {
 506                 rc = next->do_index_ops->dio_delete(env, next,
 507                                                     (struct dt_key *)name,
 508                                                     handle, capa);
 509         } else
 510                 rc = -ENOTDIR;
 511
 512         RETURN(rc);
 513 }
 514
 515 static int __mdd_index_insert_only(const struct lu_env *env,
 516                                    struct mdd_object *pobj,
 517                                    const struct lu_fid *lf, const char *name,
 518                                    struct thandle *handle,
 519                                    struct lustre_capa *capa)
 520 {
 521         struct dt_object *next = mdd_object_child(pobj);
 522         int               rc;
 523         ENTRY;
 524
 525         if (dt_try_as_dir(env, next)) {
 526                 struct md_ucred  *uc = md_ucred(env);
 527
 528                 rc = next->do_index_ops->dio_insert(env, next,
 529                                                     (struct dt_rec*)lf,
 530                                                     (const struct dt_key *)name,
 531                                                     handle, capa, uc->mu_cap &
 532                                                     CFS_CAP_SYS_RESOURCE_MASK);
 533         } else {
 534                 rc = -ENOTDIR;
 535         }
 536         RETURN(rc);
 537 }
 538
 539 /* insert named index, add reference if isdir */
 540 static int __mdd_index_insert(const struct lu_env *env, struct mdd_object *pobj,
 541                               const struct lu_fid *lf, const char *name, int is_dir,
 542                               struct thandle *handle, struct lustre_capa *capa)
 543 {
 544         int               rc;
 545         ENTRY;
 546
 547         rc = __mdd_index_insert_only(env, pobj, lf, name, handle, capa);
 548         if (rc == 0 && is_dir) {
 549                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
 550                 mdo_ref_add(env, pobj, handle);
 551                 mdd_write_unlock(env, pobj);
 552         }
 553         RETURN(rc);
 554 }
 555
 556 /* delete named index, drop reference if isdir */
 557 static int __mdd_index_delete(const struct lu_env *env, struct mdd_object *pobj,
 558                               const char *name, int is_dir, struct thandle *handle,
 559                               struct lustre_capa *capa)
 560 {
 561         int               rc;
 562         ENTRY;
 563
 564         rc = __mdd_index_delete_only(env, pobj, name, handle, capa);
 565         if (rc == 0 && is_dir) {
 566                 mdd_write_lock(env, pobj, MOR_TGT_PARENT);
 567                 mdo_ref_del(env, pobj, handle);
 568                 mdd_write_unlock(env, pobj);
 569         }
 570
 571         RETURN(rc);
 572 }
 573
 574 int mdd_declare_llog_record(const struct lu_env *env, struct mdd_device *mdd,
 575                             int reclen, struct thandle *handle)
 576 {
 577         int rc;
 578
 579         /* XXX: this is a temporary solution to declare llog changes
 580          *      will be fixed in 2.3 with new llog implementation */
 581
 582         LASSERT(mdd->mdd_capa);
 583
 584         /* record itself */
 585         rc = dt_declare_record_write(env, mdd->mdd_capa, reclen, 0, handle);
 586         if (rc)
 587                 return rc;
 588
 589         /* header will be updated as well */
 590         rc = dt_declare_record_write(env, mdd->mdd_capa, LLOG_CHUNK_SIZE,
 591                                      0, handle);
 592         if (rc)
 593                 return rc;
 594
 595         /* also we should be able to create new plain log */
 596         rc = dt_declare_create(env, mdd->mdd_capa, NULL, NULL, NULL, handle);
 597         if (rc)
 598                 return rc;
 599
 600         /* new record referencing new plain llog */
 601         rc = dt_declare_record_write(env, mdd->mdd_capa,
 602                                      sizeof(struct llog_logid_rec), 0, handle);
 603         if (rc)
 604                 return rc;
 605
 606         /* catalog's header will be updated as well */
 607         rc = dt_declare_record_write(env, mdd->mdd_capa, LLOG_CHUNK_SIZE,
 608                                      0, handle);
 609
 610         return rc;
 611 }
 612
 613 int mdd_declare_changelog_store(const struct lu_env *env,
 614                                 struct mdd_device *mdd,
 615                                 const struct lu_name *fname,
 616                                 struct thandle *handle)
 617 {
 618         int reclen;
 619
 620         /* Not recording */
 621         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 622                 return 0;
 623
 624         /* we'll be writing payload + llog header */
 625         reclen = sizeof(struct llog_changelog_rec);
 626         if (fname)
 627                 reclen += llog_data_len(fname->ln_namelen);
 628
 629         return mdd_declare_llog_record(env, mdd, reclen, handle);
 630 }
 631
 632 /** Store a namespace change changelog record
 633  * If this fails, we must fail the whole transaction; we don't
 634  * want the change to commit without the log entry.
 635  * \param target - mdd_object of change
 636  * \param parent - parent dir/object
 637  * \param tf - target lu_fid, overrides fid of \a target if this is non-null
 638  * \param tname - target name string
 639  * \param handle - transacion handle
 640  */
 641 static int mdd_changelog_ns_store(const struct lu_env  *env,
 642                                   struct mdd_device    *mdd,
 643                                   enum changelog_rec_type type,
 644                                   int flags,
 645                                   struct mdd_object    *target,
 646                                   struct mdd_object    *parent,
 647                                   const struct lu_fid  *tf,
 648                                   const struct lu_name *tname,
 649                                   struct thandle *handle)
 650 {
 651         const struct lu_fid *tfid;
 652         const struct lu_fid *tpfid = mdo2fid(parent);
 653         struct llog_changelog_rec *rec;
 654         struct lu_buf *buf;
 655         int reclen;
 656         int rc;
 657         ENTRY;
 658
 659         /* Not recording */
 660         if (!(mdd->mdd_cl.mc_flags & CLM_ON))
 661                 RETURN(0);
 662         if ((mdd->mdd_cl.mc_mask & (1 << type)) == 0)
 663                 RETURN(0);
 664
 665         LASSERT(parent != NULL);
 666         LASSERT(tname != NULL);
 667         LASSERT(handle != NULL);
 668
 669         /* target */
 670         reclen = llog_data_len(sizeof(*rec) + tname->ln_namelen);
 671         buf = mdd_buf_alloc(env, reclen);
 672         if (buf->lb_buf == NULL)
 673                 RETURN(-ENOMEM);
 674         rec = (struct llog_changelog_rec *)buf->lb_buf;
 675
 676         rec->cr.cr_flags = CLF_VERSION | (CLF_FLAGMASK & flags);
 677         rec->cr.cr_type = (__u32)type;
 678         tfid = tf ? tf : mdo2fid(target);
 679         rec->cr.cr_tfid = *tfid;
 680         rec->cr.cr_pfid = *tpfid;
 681         rec->cr.cr_namelen = tname->ln_namelen;
 682         memcpy(rec->cr.cr_name, tname->ln_name, rec->cr.cr_namelen);
 683         if (likely(target))
 684                 target->mod_cltime = cfs_time_current_64();
 685
 686         rc = mdd_changelog_llog_write(mdd, rec, handle);
 687         if (rc < 0) {
 688                 CERROR("changelog failed: rc=%d, op%d %s c"DFID" p"DFID"\n",
 689                        rc, type, tname->ln_name, PFID(tfid), PFID(tpfid));
 690                 return -EFAULT;
 691         }
 692
 693         return 0;
 694 }
 695
 696 static int mdd_declare_link(const struct lu_env *env,
 697                             struct mdd_device *mdd,
 698                             struct mdd_object *p,
 699                             struct mdd_object *c,
 700                             const struct lu_name *name,
 701                             struct thandle *handle)
 702 {
 703         int rc;
 704
 705         rc = mdo_declare_index_insert(env, p, mdo2fid(c), name->ln_name,handle);
 706         if (rc)
 707                 return rc;
 708
 709         rc = mdo_declare_ref_add(env, c, handle);
 710         if (rc)
 711                 return rc;
 712
 713         rc = mdo_declare_attr_set(env, p, NULL, handle);
 714         if (rc)
 715                 return rc;
 716
 717         rc = mdo_declare_attr_set(env, c, NULL, handle);
 718         if (rc)
 719                 return rc;
 720
 721         rc = mdd_declare_links_add(env, c, handle);
 722         if (rc)
 723                 return rc;
 724
 725         rc = mdd_declare_changelog_store(env, mdd, name, handle);
 726
 727         return rc;
 728 }
 729
 730 static int mdd_link(const struct lu_env *env, struct md_object *tgt_obj,
 731                     struct md_object *src_obj, const struct lu_name *lname,
 732                     struct md_attr *ma)
 733 {
 734         const char *name = lname->ln_name;
 735         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
 736         struct mdd_object *mdd_tobj = md2mdd_obj(tgt_obj);
 737         struct mdd_object *mdd_sobj = md2mdd_obj(src_obj);
 738         struct mdd_device *mdd = mdo2mdd(src_obj);
 739         struct dynlock_handle *dlh;
 740         struct thandle *handle;
 741 #ifdef HAVE_QUOTA_SUPPORT
 742         struct obd_device *obd = mdd->mdd_obd_dev;
 743         struct obd_export *exp = md_quota(env)->mq_exp;
 744         struct mds_obd *mds = &obd->u.mds;
 745         unsigned int qids[MAXQUOTAS] = { 0, 0 };
 746         int quota_opc = 0, rec_pending[MAXQUOTAS] = { 0, 0 };
 747 #endif
 748         int rc;
 749         ENTRY;
 750
 751 #ifdef HAVE_QUOTA_SUPPORT
 752         if (mds->mds_quota) {
 753                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
 754
 755                 rc = mdd_la_get(env, mdd_tobj, la_tmp, BYPASS_CAPA);
 756                 if (!rc) {
 757                         void *data = NULL;
 758                         mdd_data_get(env, mdd_tobj, &data);
 759                         quota_opc = FSFILT_OP_LINK;
 760                         mdd_quota_wrapper(la_tmp, qids);
 761                         /* get block quota for parent */
 762                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
 763                                         qids, rec_pending, 1, NULL,
 764                                         LQUOTA_FLAGS_BLK, data, 1);
 765                 }
 766         }
 767 #endif
 768
 769         handle = mdd_trans_create(env, mdd);
 770         if (IS_ERR(handle))
 771                 GOTO(out_pending, rc = PTR_ERR(handle));
 772
 773         rc = mdd_declare_link(env, mdd, mdd_tobj, mdd_sobj, lname, handle);
 774         if (rc)
 775                 GOTO(stop, rc);
 776
 777         rc = mdd_trans_start(env, mdd, handle);
 778         if (rc)
 779                 GOTO(stop, rc);
 780
 781         dlh = mdd_pdo_write_lock(env, mdd_tobj, name, MOR_TGT_CHILD);
 782         if (dlh == NULL)
 783                 GOTO(out_trans, rc = -ENOMEM);
 784         mdd_write_lock(env, mdd_sobj, MOR_TGT_CHILD);
 785
 786         rc = mdd_link_sanity_check(env, mdd_tobj, lname, mdd_sobj);
 787         if (rc)
 788                 GOTO(out_unlock, rc);
 789
 790         rc = __mdd_index_insert_only(env, mdd_tobj, mdo2fid(mdd_sobj),
 791                                      name, handle,
 792                                      mdd_object_capa(env, mdd_tobj));
 793         if (rc)
 794                 GOTO(out_unlock, rc);
 795
 796         mdo_ref_add(env, mdd_sobj, handle);
 797
 798         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
 799         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
 800
 801         la->la_valid = LA_CTIME | LA_MTIME;
 802         rc = mdd_attr_check_set_internal_locked(env, mdd_tobj, la, handle, 0);
 803         if (rc)
 804                 GOTO(out_unlock, rc);
 805
 806         la->la_valid = LA_CTIME;
 807         rc = mdd_attr_check_set_internal(env, mdd_sobj, la, handle, 0);
 808         if (rc == 0) {
 809                 mdd_links_add(env, mdd_sobj,
 810                               mdo2fid(mdd_tobj), lname, handle, 0);
 811         }
 812
 813         EXIT;
 814 out_unlock:
 815         mdd_write_unlock(env, mdd_sobj);
 816         mdd_pdo_write_unlock(env, mdd_tobj, dlh);
 817 out_trans:
 818         if (rc == 0)
 819                 rc = mdd_changelog_ns_store(env, mdd, CL_HARDLINK, 0, mdd_sobj,
 820                                             mdd_tobj, NULL, lname, handle);
 821 stop:
 822         mdd_trans_stop(env, mdd, rc, handle);
 823 out_pending:
 824 #ifdef HAVE_QUOTA_SUPPORT
 825         if (quota_opc) {
 826                 lquota_pending_commit(mds_quota_interface_ref, obd,
 827                                       qids, rec_pending, 1);
 828                 /* Trigger dqacq for the parent owner. If failed,
 829                  * the next call for lquota_chkquota will process it. */
 830                 lquota_adjust(mds_quota_interface_ref, obd, 0, qids, rc,
 831                               quota_opc);
 832         }
 833 #endif
 834         return rc;
 835 }
 836
 837 int mdd_declare_finish_unlink(const struct lu_env *env,
 838                               struct mdd_object *obj,
 839                               struct md_attr *ma,
 840                               struct thandle *handle)
 841 {
 842         int rc;
 843
 844         rc = orph_declare_index_insert(env, obj, handle);
 845         if (rc)
 846                 return rc;
 847
 848         return mdd_declare_object_kill(env, obj, ma, handle);
 849 }
 850
 851 /* caller should take a lock before calling */
 852 int mdd_finish_unlink(const struct lu_env *env,
 853                       struct mdd_object *obj, struct md_attr *ma,
 854                       struct thandle *th)
 855 {
 856         int rc;
 857         int reset = 1;
 858         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
 859         ENTRY;
 860
 861         LASSERT(mdd_write_locked(env, obj) != 0);
 862
 863         /* read HSM flags, needed to set changelogs flags */
 864         ma->ma_need = MA_HSM | MA_INODE;
 865         rc = mdd_attr_get_internal(env, obj, ma);
 866         if (rc == 0 && (ma->ma_attr.la_nlink == 0 || is_dir)) {
 867                 obj->mod_flags |= DEAD_OBJ;
 868                 /* add new orphan and the object
 869                  * will be deleted during mdd_close() */
 870                 if (obj->mod_count) {
 871                         rc = __mdd_orphan_add(env, obj, th);
 872                         if (rc == 0)
 873                                 CDEBUG(D_HA, "Object "DFID" is inserted into "
 874                                         "orphan list, open count = %d\n",
 875                                         PFID(mdd_object_fid(obj)),
 876                                         obj->mod_count);
 877                         else
 878                                 CERROR("Object "DFID" fail to be an orphan, "
 879                                        "open count = %d, maybe cause failed "
 880                                        "open replay\n",
 881                                         PFID(mdd_object_fid(obj)),
 882                                         obj->mod_count);
 883                 } else {
 884                         rc = mdd_object_kill(env, obj, ma, th);
 885                         if (rc == 0)
 886                                 reset = 0;
 887                 }
 888
 889                 /* get the i_nlink */
 890                 ma->ma_need = MA_INODE;
 891                 rc = mdd_attr_get_internal(env, obj, ma);
 892         }
 893         if (reset)
 894                 ma->ma_valid &= ~(MA_LOV | MA_COOKIE);
 895
 896         RETURN(rc);
 897 }
 898
 899 /*
 900  * pobj maybe NULL
 901  * has mdd_write_lock on cobj already, but not on pobj yet
 902  */
 903 int mdd_unlink_sanity_check(const struct lu_env *env, struct mdd_object *pobj,
 904                             struct mdd_object *cobj, struct md_attr *ma)
 905 {
 906         int rc;
 907         ENTRY;
 908
 909         rc = mdd_may_delete(env, pobj, cobj, ma, 1, 1);
 910
 911         RETURN(rc);
 912 }
 913
 914 static int mdd_declare_unlink(const struct lu_env *env, struct mdd_device *mdd,
 915                               struct mdd_object *p, struct mdd_object *c,
 916                               const struct lu_name *name, struct md_attr *ma,
 917                               struct thandle *handle)
 918 {
 919         int rc;
 920
 921         rc = mdo_declare_index_delete(env, p, name->ln_name, handle);
 922         if (rc)
 923                 return rc;
 924
 925         rc = mdo_declare_ref_del(env, p, handle);
 926         if (rc)
 927                 return rc;
 928
 929         rc = mdo_declare_ref_del(env, c, handle);
 930         if (rc)
 931                 return rc;
 932
 933         rc = mdo_declare_ref_del(env, c, handle);
 934         if (rc)
 935                 return rc;
 936
 937         rc = mdo_declare_attr_set(env, p, NULL, handle);
 938         if (rc)
 939                 return rc;
 940
 941         rc = mdo_declare_attr_set(env, c, NULL, handle);
 942         if (rc)
 943                 return rc;
 944
 945         rc = mdd_declare_finish_unlink(env, c, ma, handle);
 946         if (rc)
 947                 return rc;
 948
 949         rc = mdd_declare_links_add(env, c, handle);
 950         if (rc)
 951                 return rc;
 952
 953         rc = mdd_declare_changelog_store(env, mdd, name, handle);
 954
 955         return rc;
 956 }
 957
 958 static int mdd_unlink(const struct lu_env *env, struct md_object *pobj,
 959                       struct md_object *cobj, const struct lu_name *lname,
 960                       struct md_attr *ma)
 961 {
 962         const char *name = lname->ln_name;
 963         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
 964         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
 965         struct mdd_object *mdd_cobj = md2mdd_obj(cobj);
 966         struct mdd_device *mdd = mdo2mdd(pobj);
 967         struct dynlock_handle *dlh;
 968         struct thandle    *handle;
 969 #ifdef HAVE_QUOTA_SUPPORT
 970         struct obd_device *obd = mdd->mdd_obd_dev;
 971         struct mds_obd *mds = &obd->u.mds;
 972         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
 973         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
 974         int quota_opc = 0;
 975 #endif
 976         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
 977         int rc;
 978         ENTRY;
 979
 980         if (mdd_object_exists(mdd_cobj) <= 0)
 981                 RETURN(-ENOENT);
 982
 983         handle = mdd_trans_create(env, mdd);
 984         if (IS_ERR(handle))
 985                 RETURN(PTR_ERR(handle));
 986
 987         rc = mdd_declare_unlink(env, mdd, mdd_pobj, mdd_cobj,
 988                                 lname, ma, handle);
 989         if (rc)
 990                 GOTO(stop, rc);
 991
 992         rc = mdd_trans_start(env, mdd, handle);
 993         if (rc)
 994                 GOTO(stop, rc);
 995
 996         dlh = mdd_pdo_write_lock(env, mdd_pobj, name, MOR_TGT_PARENT);
 997         if (dlh == NULL)
 998                 GOTO(out_trans, rc = -ENOMEM);
 999         mdd_write_lock(env, mdd_cobj, MOR_TGT_CHILD);
1000
1001         rc = mdd_unlink_sanity_check(env, mdd_pobj, mdd_cobj, ma);
1002         if (rc)
1003                 GOTO(cleanup, rc);
1004
1005         rc = __mdd_index_delete(env, mdd_pobj, name, is_dir, handle,
1006                                 mdd_object_capa(env, mdd_pobj));
1007         if (rc)
1008                 GOTO(cleanup, rc);
1009
1010         mdo_ref_del(env, mdd_cobj, handle);
1011         if (is_dir)
1012                 /* unlink dot */
1013                 mdo_ref_del(env, mdd_cobj, handle);
1014
1015         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1016         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1017
1018         la->la_valid = LA_CTIME | LA_MTIME;
1019         rc = mdd_attr_check_set_internal_locked(env, mdd_pobj, la, handle, 0);
1020         if (rc)
1021                 GOTO(cleanup, rc);
1022
1023         if (ma->ma_attr.la_nlink > 0 || mdd_cobj->mod_count > 0) {
1024                 /* update ctime of an unlinked file only if it is still
1025                  * opened or a link still exists */
1026                 la->la_valid = LA_CTIME;
1027                 rc = mdd_attr_check_set_internal(env, mdd_cobj, la, handle, 0);
1028                 if (rc)
1029                         GOTO(cleanup, rc);
1030         }
1031
1032         rc = mdd_finish_unlink(env, mdd_cobj, ma, handle);
1033 #ifdef HAVE_QUOTA_SUPPORT
1034         if (mds->mds_quota && ma->ma_valid & MA_INODE &&
1035             ma->ma_attr.la_nlink == 0) {
1036                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1037
1038                 rc = mdd_la_get(env, mdd_pobj, la_tmp, BYPASS_CAPA);
1039                 if (!rc) {
1040                         mdd_quota_wrapper(la_tmp, qpids);
1041                         if (mdd_cobj->mod_count == 0) {
1042                                 quota_opc = FSFILT_OP_UNLINK;
1043                                 mdd_quota_wrapper(&ma->ma_attr, qcids);
1044                         } else {
1045                                 quota_opc = FSFILT_OP_UNLINK_PARTIAL_PARENT;
1046                         }
1047                 }
1048         }
1049 #endif
1050         if (!is_dir)
1051                 /* old files may not have link ea; ignore errors */
1052                 mdd_links_rename(env, mdd_cobj, mdo2fid(mdd_pobj),
1053                                  lname, NULL, NULL, handle);
1054
1055         EXIT;
1056 cleanup:
1057         mdd_write_unlock(env, mdd_cobj);
1058         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
1059 out_trans:
1060         if (rc == 0) {
1061                 int cl_flags;
1062
1063                 cl_flags = (ma->ma_attr.la_nlink == 0) ? CLF_UNLINK_LAST : 0;
1064                 if ((ma->ma_valid & MA_HSM) &&
1065                     (ma->ma_hsm.mh_flags & HS_EXISTS))
1066                         cl_flags |= CLF_UNLINK_HSM_EXISTS;
1067
1068                 rc = mdd_changelog_ns_store(env, mdd,
1069                          is_dir ? CL_RMDIR : CL_UNLINK, cl_flags,
1070                          mdd_cobj, mdd_pobj, NULL, lname, handle);
1071         }
1072
1073 stop:
1074         mdd_trans_stop(env, mdd, rc, handle);
1075 #ifdef HAVE_QUOTA_SUPPORT
1076         if (quota_opc)
1077                 /* Trigger dqrel on the owner of child and parent. If failed,
1078                  * the next call for lquota_chkquota will process it. */
1079                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
1080                               quota_opc);
1081 #endif
1082         return rc;
1083 }
1084
1085 /* has not lock on pobj yet */
1086 static int mdd_ni_sanity_check(const struct lu_env *env,
1087                                struct md_object *pobj,
1088                                const struct md_attr *ma)
1089 {
1090         struct mdd_object *obj = md2mdd_obj(pobj);
1091         int rc;
1092         ENTRY;
1093
1094         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
1095                 RETURN(0);
1096
1097         rc = mdd_may_create(env, obj, NULL, 1, S_ISDIR(ma->ma_attr.la_mode));
1098
1099         RETURN(rc);
1100 }
1101
1102 /*
1103  * Partial operation.
1104  */
1105 static int mdd_name_insert(const struct lu_env *env,
1106                            struct md_object *pobj,
1107                            const struct lu_name *lname,
1108                            const struct lu_fid *fid,
1109                            const struct md_attr *ma)
1110 {
1111         const char *name = lname->ln_name;
1112         struct lu_attr   *la = &mdd_env_info(env)->mti_la_for_fix;
1113         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1114         struct mdd_device *mdd = mdo2mdd(pobj);
1115         struct dynlock_handle *dlh;
1116         struct thandle *handle;
1117         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1118 #ifdef HAVE_QUOTA_SUPPORT
1119         struct md_ucred *uc = md_ucred(env);
1120         struct obd_device *obd = mdd->mdd_obd_dev;
1121         struct obd_export *exp = md_quota(env)->mq_exp;
1122         struct mds_obd *mds = &obd->u.mds;
1123         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1124         int quota_opc = 0, rec_pending[MAXQUOTAS] = { 0, 0 };
1125         cfs_cap_t save = uc->mu_cap;
1126 #endif
1127         int rc;
1128         ENTRY;
1129
1130         /* XXX: this code won't be used ever:
1131          * DNE uses slightly different approach */
1132         LBUG();
1133
1134 #ifdef HAVE_QUOTA_SUPPORT
1135         if (mds->mds_quota) {
1136                 if (!(ma->ma_attr_flags & MDS_QUOTA_IGNORE)) {
1137                         struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1138
1139                         rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1140                         if (!rc) {
1141                                 void *data = NULL;
1142                                 mdd_data_get(env, mdd_obj, &data);
1143                                 quota_opc = FSFILT_OP_LINK;
1144                                 mdd_quota_wrapper(la_tmp, qids);
1145                                 /* get block quota for parent */
1146                                 lquota_chkquota(mds_quota_interface_ref, obd,
1147                                                 exp, qids, rec_pending, 1, NULL,
1148                                                 LQUOTA_FLAGS_BLK, data, 1);
1149                         }
1150                 } else {
1151                         uc->mu_cap |= CFS_CAP_SYS_RESOURCE_MASK;
1152                 }
1153         }
1154 #endif
1155         handle = mdd_trans_create(env, mdo2mdd(pobj));
1156         if (IS_ERR(handle))
1157                 GOTO(out_pending, rc = PTR_ERR(handle));
1158
1159         rc = mdd_trans_start(env, mdo2mdd(pobj), handle);
1160
1161         dlh = mdd_pdo_write_lock(env, mdd_obj, name, MOR_TGT_PARENT);
1162         if (dlh == NULL)
1163                 GOTO(out_trans, rc = -ENOMEM);
1164
1165         rc = mdd_ni_sanity_check(env, pobj, ma);
1166         if (rc)
1167                 GOTO(out_unlock, rc);
1168
1169         rc = __mdd_index_insert(env, mdd_obj, fid, name, is_dir,
1170                                 handle, BYPASS_CAPA);
1171         if (rc)
1172                 GOTO(out_unlock, rc);
1173
1174         /*
1175          * For some case, no need update obj's ctime (LA_CTIME is not set),
1176          * e.g. split_dir.
1177          * For other cases, update obj's ctime (LA_CTIME is set),
1178          * e.g. cmr_link.
1179          */
1180         if (ma->ma_attr.la_valid & LA_CTIME) {
1181                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1182                 la->la_valid = LA_CTIME | LA_MTIME;
1183                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la,
1184                                                         handle, 0);
1185         }
1186         EXIT;
1187 out_unlock:
1188         mdd_pdo_write_unlock(env, mdd_obj, dlh);
1189 out_trans:
1190         mdd_trans_stop(env, mdo2mdd(pobj), rc, handle);
1191 out_pending:
1192 #ifdef HAVE_QUOTA_SUPPORT
1193         if (mds->mds_quota) {
1194                 if (quota_opc) {
1195                         lquota_pending_commit(mds_quota_interface_ref,
1196                                               obd, qids, rec_pending, 1);
1197                         /* Trigger dqacq for the parent owner. If failed,
1198                          * the next call for lquota_chkquota will process it*/
1199                         lquota_adjust(mds_quota_interface_ref, obd, 0, qids,
1200                                       rc, quota_opc);
1201                 } else {
1202                         uc->mu_cap = save;
1203                 }
1204         }
1205 #endif
1206         return rc;
1207 }
1208
1209 /* has not lock on pobj yet */
1210 static int mdd_nr_sanity_check(const struct lu_env *env,
1211                                struct md_object *pobj,
1212                                const struct md_attr *ma)
1213 {
1214         struct mdd_object *obj = md2mdd_obj(pobj);
1215         int rc;
1216         ENTRY;
1217
1218         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
1219                 RETURN(0);
1220
1221         rc = mdd_may_unlink(env, obj, ma);
1222
1223         RETURN(rc);
1224 }
1225
1226 /*
1227  * Partial operation.
1228  */
1229 static int mdd_name_remove(const struct lu_env *env,
1230                            struct md_object *pobj,
1231                            const struct lu_name *lname,
1232                            const struct md_attr *ma)
1233 {
1234         const char *name = lname->ln_name;
1235         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1236         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1237         struct mdd_device *mdd = mdo2mdd(pobj);
1238         struct dynlock_handle *dlh;
1239         struct thandle *handle;
1240         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1241 #ifdef HAVE_QUOTA_SUPPORT
1242         struct obd_device *obd = mdd->mdd_obd_dev;
1243         struct mds_obd *mds = &obd->u.mds;
1244         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1245         int quota_opc = 0;
1246 #endif
1247         int rc;
1248         ENTRY;
1249
1250         /* XXX: this code won't be used ever:
1251          * DNE uses slightly different approach */
1252         LBUG();
1253
1254 #ifdef HAVE_QUOTA_SUPPORT
1255         if (mds->mds_quota) {
1256                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1257
1258                 rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1259                 if (!rc) {
1260                         quota_opc = FSFILT_OP_UNLINK_PARTIAL_PARENT;
1261                         mdd_quota_wrapper(la_tmp, qids);
1262                 }
1263         }
1264 #endif
1265         handle = mdd_trans_create(env, mdd);
1266         if (IS_ERR(handle))
1267                 GOTO(out_pending, rc = PTR_ERR(handle));
1268
1269         rc = mdd_trans_start(env, mdd, handle);
1270
1271         dlh = mdd_pdo_write_lock(env, mdd_obj, name, MOR_TGT_PARENT);
1272         if (dlh == NULL)
1273                 GOTO(out_trans, rc = -ENOMEM);
1274
1275         rc = mdd_nr_sanity_check(env, pobj, ma);
1276         if (rc)
1277                 GOTO(out_unlock, rc);
1278
1279         rc = __mdd_index_delete(env, mdd_obj, name, is_dir,
1280                                 handle, BYPASS_CAPA);
1281         if (rc)
1282                 GOTO(out_unlock, rc);
1283
1284         /*
1285          * For some case, no need update obj's ctime (LA_CTIME is not set),
1286          * e.g. split_dir.
1287          * For other cases, update obj's ctime (LA_CTIME is set),
1288          * e.g. cmr_unlink.
1289          */
1290         if (ma->ma_attr.la_valid & LA_CTIME) {
1291                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1292                 la->la_valid = LA_CTIME | LA_MTIME;
1293                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la,
1294                                                         handle, 0);
1295         }
1296         EXIT;
1297 out_unlock:
1298         mdd_pdo_write_unlock(env, mdd_obj, dlh);
1299 out_trans:
1300         mdd_trans_stop(env, mdd, rc, handle);
1301 out_pending:
1302 #ifdef HAVE_QUOTA_SUPPORT
1303         /* Trigger dqrel for the parent owner.
1304          * If failed, the next call for lquota_chkquota will process it. */
1305         if (quota_opc)
1306                 lquota_adjust(mds_quota_interface_ref, obd, 0, qids, rc,
1307                               quota_opc);
1308 #endif
1309         return rc;
1310 }
1311
1312 /*
1313  * tobj maybe NULL
1314  * has mdd_write_lock on tobj alreay, but not on tgt_pobj yet
1315  */
1316 static int mdd_rt_sanity_check(const struct lu_env *env,
1317                                struct mdd_object *tgt_pobj,
1318                                struct mdd_object *tobj,
1319                                struct md_attr *ma)
1320 {
1321         int rc;
1322         ENTRY;
1323
1324         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
1325                 RETURN(0);
1326
1327         /* XXX: for mdd_rename_tgt, "tobj == NULL" does not mean tobj not
1328          * exist. In fact, tobj must exist, otherwise the call trace will be:
1329          * mdt_reint_rename_tgt -> mdo_name_insert -> ... -> mdd_name_insert.
1330          * When get here, tobj must be NOT NULL, the other case has been
1331          * processed in cmr_rename_tgt before mdd_rename_tgt and enable
1332          * MDS_PERM_BYPASS.
1333          * So check may_delete, but not check nlink of tgt_pobj. */
1334
1335         rc = mdd_may_delete(env, tgt_pobj, tobj, ma, 1, 1);
1336
1337         RETURN(rc);
1338 }
1339
1340 /* Partial rename op on slave MDD */
1341 static int mdd_rename_tgt(const struct lu_env *env,
1342                           struct md_object *pobj, struct md_object *tobj,
1343                           const struct lu_fid *lf, const struct lu_name *lname,
1344                           struct md_attr *ma)
1345 {
1346         const char *name = lname->ln_name;
1347         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1348         struct mdd_object *mdd_tpobj = md2mdd_obj(pobj);
1349         struct mdd_object *mdd_tobj = md2mdd_obj(tobj);
1350         struct mdd_device *mdd = mdo2mdd(pobj);
1351         struct dynlock_handle *dlh;
1352         struct thandle *handle;
1353 #ifdef HAVE_QUOTA_SUPPORT
1354         struct obd_device *obd = mdd->mdd_obd_dev;
1355         struct obd_export *exp = md_quota(env)->mq_exp;
1356         struct mds_obd *mds = &obd->u.mds;
1357         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1358         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1359         int quota_copc = 0, quota_popc = 0;
1360         int rec_pending[MAXQUOTAS] = { 0, 0 };
1361 #endif
1362         int rc;
1363         ENTRY;
1364
1365         /* XXX: this code won't be used ever:
1366          * DNE uses slightly different approach */
1367         LBUG();
1368
1369 #ifdef HAVE_QUOTA_SUPPORT
1370         if (mds->mds_quota && !tobj) {
1371                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1372
1373                 rc = mdd_la_get(env, mdd_tpobj, la_tmp, BYPASS_CAPA);
1374                 if (!rc) {
1375                         void *data = NULL;
1376                         mdd_data_get(env, mdd_tpobj, &data);
1377                         quota_popc = FSFILT_OP_LINK;
1378                         mdd_quota_wrapper(la_tmp, qpids);
1379                         /* get block quota for target parent */
1380                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1381                                         qpids, rec_pending, 1, NULL,
1382                                         LQUOTA_FLAGS_BLK, data, 1);
1383                 }
1384         }
1385 #endif
1386         handle = mdd_trans_create(env, mdd);
1387         if (IS_ERR(handle))
1388                 GOTO(out_pending, rc = PTR_ERR(handle));
1389
1390         rc = mdd_trans_start(env, mdd, handle);
1391
1392         dlh = mdd_pdo_write_lock(env, mdd_tpobj, name, MOR_TGT_PARENT);
1393         if (dlh == NULL)
1394                 GOTO(out_trans, rc = -ENOMEM);
1395         if (tobj)
1396                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
1397
1398         rc = mdd_rt_sanity_check(env, mdd_tpobj, mdd_tobj, ma);
1399         if (rc)
1400                 GOTO(cleanup, rc);
1401
1402         /*
1403          * If rename_tgt is called then we should just re-insert name with
1404          * correct fid, no need to dec/inc parent nlink if obj is dir.
1405          */
1406         rc = __mdd_index_delete(env, mdd_tpobj, name, 0, handle, BYPASS_CAPA);
1407         if (rc)
1408                 GOTO(cleanup, rc);
1409
1410         rc = __mdd_index_insert_only(env, mdd_tpobj, lf, name, handle,
1411                                      BYPASS_CAPA);
1412         if (rc)
1413                 GOTO(cleanup, rc);
1414
1415         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1416         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1417
1418         la->la_valid = LA_CTIME | LA_MTIME;
1419         rc = mdd_attr_check_set_internal_locked(env, mdd_tpobj, la, handle, 0);
1420         if (rc)
1421                 GOTO(cleanup, rc);
1422
1423         /*
1424          * For tobj is remote case cmm layer has processed
1425          * and pass NULL tobj to here. So when tobj is NOT NULL,
1426          * it must be local one.
1427          */
1428         if (tobj && mdd_object_exists(mdd_tobj)) {
1429                 mdo_ref_del(env, mdd_tobj, handle);
1430
1431                 /* Remove dot reference. */
1432                 if (S_ISDIR(ma->ma_attr.la_mode))
1433                         mdo_ref_del(env, mdd_tobj, handle);
1434
1435                 la->la_valid = LA_CTIME;
1436                 rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
1437                 if (rc)
1438                         GOTO(cleanup, rc);
1439
1440                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
1441                 if (rc)
1442                         GOTO(cleanup, rc);
1443
1444 #ifdef HAVE_QUOTA_SUPPORT
1445                 if (mds->mds_quota && ma->ma_valid & MA_INODE &&
1446                     ma->ma_attr.la_nlink == 0 && mdd_tobj->mod_count == 0) {
1447                         quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
1448                         mdd_quota_wrapper(&ma->ma_attr, qcids);
1449                 }
1450 #endif
1451         }
1452         EXIT;
1453 cleanup:
1454         if (tobj)
1455                 mdd_write_unlock(env, mdd_tobj);
1456         mdd_pdo_write_unlock(env, mdd_tpobj, dlh);
1457 out_trans:
1458         if (rc == 0)
1459                 /* Bare EXT record with no RENAME in front of it signifies
1460                    a partial slave op */
1461                 rc = mdd_changelog_ns_store(env, mdd, CL_EXT, 0, mdd_tobj,
1462                                             mdd_tpobj, NULL, lname, handle);
1463
1464         mdd_trans_stop(env, mdd, rc, handle);
1465 out_pending:
1466 #ifdef HAVE_QUOTA_SUPPORT
1467         if (mds->mds_quota) {
1468                 if (quota_popc)
1469                         lquota_pending_commit(mds_quota_interface_ref, obd,
1470                                               qpids, rec_pending, 1);
1471
1472                 if (quota_copc)
1473                         /* Trigger dqrel on the target owner of child.
1474                          * If failed, the next call for lquota_chkquota
1475                          * will process it. */
1476                         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids,
1477                                       rc, quota_copc);
1478         }
1479 #endif
1480         return rc;
1481 }
1482
1483 /*
1484  * The permission has been checked when obj created, no need check again.
1485  */
1486 static int mdd_cd_sanity_check(const struct lu_env *env,
1487                                struct mdd_object *obj)
1488 {
1489         ENTRY;
1490
1491         /* EEXIST check */
1492         if (!obj || mdd_is_dead_obj(obj))
1493                 RETURN(-ENOENT);
1494
1495         RETURN(0);
1496
1497 }
1498
1499 static int mdd_declare_create_data(const struct lu_env *env,
1500                                    struct mdd_device *mdd,
1501                                    struct mdd_object *obj,
1502                                    int lmm_size,
1503                                    struct thandle *handle)
1504 {
1505         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1506         int            rc;
1507
1508         buf->lb_buf = NULL;
1509         buf->lb_len = lmm_size;
1510         rc = mdo_declare_xattr_set(env, obj, buf, XATTR_NAME_LOV,
1511                                    0, handle);
1512         if (rc)
1513                 return rc;
1514
1515         rc = mdd_declare_lov_objid_update(env, mdd, handle);
1516
1517         return rc;
1518 }
1519
1520 static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
1521                            struct md_object *cobj, const struct md_op_spec *spec,
1522                            struct md_attr *ma)
1523 {
1524         struct mdd_device *mdd = mdo2mdd(cobj);
1525         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1526         struct mdd_object *son = md2mdd_obj(cobj);
1527         struct lov_mds_md *lmm = NULL;
1528         int                lmm_size = 0;
1529         struct thandle    *handle;
1530         int                rc;
1531         ENTRY;
1532
1533         rc = mdd_cd_sanity_check(env, son);
1534         if (rc)
1535                 RETURN(rc);
1536
1537         if (!md_should_create(spec->sp_cr_flags))
1538                 RETURN(0);
1539         lmm_size = ma->ma_lmm_size;
1540
1541         rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size, spec, ma);
1542         if (rc)
1543                 RETURN(rc);
1544
1545         handle = mdd_trans_create(env, mdd);
1546         if (IS_ERR(handle))
1547                 GOTO(out_free, rc = PTR_ERR(handle));
1548
1549         rc = mdd_declare_create_data(env, mdd, son, lmm_size, handle);
1550         if (rc)
1551                 GOTO(stop, rc);
1552
1553         rc = mdd_trans_start(env, mdd, handle);
1554         if (rc)
1555                 GOTO(stop, rc);
1556
1557         /*
1558          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1559          * Should this be fixed?
1560          */
1561
1562         /* Replay creates has objects already */
1563 #if 0
1564         if (spec->no_create) {
1565                 CDEBUG(D_INFO, "we already have lov ea\n");
1566                 rc = mdd_lov_set_md(env, mdd_pobj, son,
1567                                     (struct lov_mds_md *)spec->u.sp_ea.eadata,
1568                                     spec->u.sp_ea.eadatalen, handle, 0);
1569         } else
1570 #endif
1571                 /* No need mdd_lsm_sanity_check here */
1572                 rc = mdd_lov_set_md(env, mdd_pobj, son, lmm,
1573                                     lmm_size, handle, 0);
1574
1575         if (rc == 0)
1576                rc = mdd_attr_get_internal_locked(env, son, ma);
1577
1578         /* update lov_objid data, must be before transaction stop! */
1579         if (rc == 0)
1580                 mdd_lov_objid_update(mdd, lmm);
1581
1582 stop:
1583         mdd_trans_stop(env, mdd, rc, handle);
1584 out_free:
1585         /* Finish mdd_lov_create() stuff. */
1586         /* if no_create == 0 (not replay), we free lmm allocated by
1587          * mdd_lov_create() */
1588         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
1589         RETURN(rc);
1590 }
1591
1592 /* Get fid from name and parent */
1593 static int
1594 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
1595              const struct lu_name *lname, struct lu_fid* fid, int mask)
1596 {
1597         const char          *name = lname->ln_name;
1598         const struct dt_key *key = (const struct dt_key *)name;
1599         struct mdd_object   *mdd_obj = md2mdd_obj(pobj);
1600         struct mdd_device   *m = mdo2mdd(pobj);
1601         struct dt_object    *dir = mdd_object_child(mdd_obj);
1602         int rc;
1603         ENTRY;
1604
1605         if (unlikely(mdd_is_dead_obj(mdd_obj)))
1606                 RETURN(-ESTALE);
1607
1608         rc = mdd_object_exists(mdd_obj);
1609         if (unlikely(rc == 0))
1610                 RETURN(-ESTALE);
1611         else if (unlikely(rc < 0)) {
1612                 CERROR("Object "DFID" locates on remote server\n",
1613                         PFID(mdo2fid(mdd_obj)));
1614                 RETURN(-EINVAL);
1615         }
1616
1617         /* The common filename length check. */
1618         if (unlikely(lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
1619                 RETURN(-ENAMETOOLONG);
1620
1621         rc = mdd_permission_internal_locked(env, mdd_obj, NULL, mask,
1622                                             MOR_TGT_PARENT);
1623         if (rc)
1624                 RETURN(rc);
1625
1626         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
1627                    dt_try_as_dir(env, dir))) {
1628
1629                 rc = dir->do_index_ops->dio_lookup(env, dir,
1630                                                  (struct dt_rec *)fid, key,
1631                                                  mdd_object_capa(env, mdd_obj));
1632                 if (rc > 0)
1633                         rc = 0;
1634                 else if (rc == 0)
1635                         rc = -ENOENT;
1636         } else
1637                 rc = -ENOTDIR;
1638
1639         RETURN(rc);
1640 }
1641
1642 int mdd_declare_object_initialize(const struct lu_env *env,
1643                                   struct mdd_object *child,
1644                                   struct md_attr *ma,
1645                                   struct thandle *handle)
1646 {
1647         int rc;
1648
1649         rc = mdo_declare_attr_set(env, child, &ma->ma_attr, handle);
1650         if (rc == 0 && S_ISDIR(ma->ma_attr.la_mode)) {
1651                 rc = mdo_declare_index_insert(env, child, mdo2fid(child),
1652                                 dot, handle);
1653                 if (rc == 0)
1654                         rc = mdo_declare_ref_add(env, child, handle);
1655         }
1656         if (rc == 0)
1657                 mdd_declare_links_add(env, child, handle);
1658
1659         return rc;
1660 }
1661
1662 int mdd_object_initialize(const struct lu_env *env, const struct lu_fid *pfid,
1663                           const struct lu_name *lname, struct mdd_object *child,
1664                           struct md_attr *ma, struct thandle *handle,
1665                           const struct md_op_spec *spec)
1666 {
1667         int rc;
1668         ENTRY;
1669
1670         /*
1671          * Update attributes for child.
1672          *
1673          * FIXME:
1674          *  (1) the valid bits should be converted between Lustre and Linux;
1675          *  (2) maybe, the child attributes should be set in OSD when creation.
1676          */
1677
1678         rc = mdd_attr_set_internal(env, child, &ma->ma_attr, handle, 0);
1679         if (rc != 0)
1680                 RETURN(rc);
1681
1682         if (S_ISDIR(ma->ma_attr.la_mode)) {
1683                 /* Add "." and ".." for newly created dir */
1684                 mdo_ref_add(env, child, handle);
1685                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1686                                              dot, handle, BYPASS_CAPA);
1687                 if (rc == 0)
1688                         rc = __mdd_index_insert_only(env, child, pfid,
1689                                                      dotdot, handle,
1690                                                      BYPASS_CAPA);
1691                 if (rc != 0)
1692                         mdo_ref_del(env, child, handle);
1693         }
1694         if (rc == 0)
1695                 mdd_links_add(env, child, pfid, lname, handle, 1);
1696
1697         RETURN(rc);
1698 }
1699
1700 /* has not lock on pobj yet */
1701 static int mdd_create_sanity_check(const struct lu_env *env,
1702                                    struct md_object *pobj,
1703                                    const struct lu_name *lname,
1704                                    struct md_attr *ma,
1705                                    struct md_op_spec *spec)
1706 {
1707         struct mdd_thread_info *info = mdd_env_info(env);
1708         struct lu_attr    *la        = &info->mti_la;
1709         struct lu_fid     *fid       = &info->mti_fid;
1710         struct mdd_object *obj       = md2mdd_obj(pobj);
1711         struct mdd_device *m         = mdo2mdd(pobj);
1712         int lookup                   = spec->sp_cr_lookup;
1713         int rc;
1714         ENTRY;
1715
1716         /* EEXIST check */
1717         if (mdd_is_dead_obj(obj))
1718                 RETURN(-ENOENT);
1719
1720         /*
1721          * In some cases this lookup is not needed - we know before if name
1722          * exists or not because MDT performs lookup for it.
1723          * name length check is done in lookup.
1724          */
1725         if (lookup) {
1726                 /*
1727                  * Check if the name already exist, though it will be checked in
1728                  * _index_insert also, for avoiding rolling back if exists
1729                  * _index_insert.
1730                  */
1731                 rc = __mdd_lookup_locked(env, pobj, lname, fid,
1732                                          MAY_WRITE | MAY_EXEC);
1733                 if (rc != -ENOENT)
1734                         RETURN(rc ? : -EEXIST);
1735         } else {
1736                 /*
1737                  * Check WRITE permission for the parent.
1738                  * EXEC permission have been checked
1739                  * when lookup before create already.
1740                  */
1741                 rc = mdd_permission_internal_locked(env, obj, NULL, MAY_WRITE,
1742                                                     MOR_TGT_PARENT);
1743                 if (rc)
1744                         RETURN(rc);
1745         }
1746
1747         /* sgid check */
1748         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
1749         if (rc != 0)
1750                 RETURN(rc);
1751
1752         if (la->la_mode & S_ISGID) {
1753                 ma->ma_attr.la_gid = la->la_gid;
1754                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1755                         ma->ma_attr.la_mode |= S_ISGID;
1756                         ma->ma_attr.la_valid |= LA_MODE;
1757                 }
1758         }
1759
1760         switch (ma->ma_attr.la_mode & S_IFMT) {
1761         case S_IFLNK: {
1762                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1763
1764                 if (symlen > (1 << m->mdd_dt_conf.ddp_block_shift))
1765                         RETURN(-ENAMETOOLONG);
1766                 else
1767                         RETURN(0);
1768         }
1769         case S_IFDIR:
1770         case S_IFREG:
1771         case S_IFCHR:
1772         case S_IFBLK:
1773         case S_IFIFO:
1774         case S_IFSOCK:
1775                 rc = 0;
1776                 break;
1777         default:
1778                 rc = -EINVAL;
1779                 break;
1780         }
1781         RETURN(rc);
1782 }
1783
1784 static int mdd_declare_create(const struct lu_env *env,
1785                               struct mdd_device *mdd,
1786                               struct mdd_object *p,
1787                               struct mdd_object *c,
1788                               const struct lu_name *name,
1789                               struct md_attr *ma,
1790                               int lmm_size,
1791                               struct thandle *handle,
1792                               const struct md_op_spec *spec)
1793 {
1794         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1795         int            rc = 0;
1796
1797         rc = mdd_declare_object_create_internal(env, p, c, ma, handle, spec);
1798         if (rc)
1799                 GOTO(out, rc);
1800
1801         /* if dir, then can inherit default ACl */
1802         buf->lb_buf = NULL;
1803         buf->lb_len = lmm_size;
1804         if (S_ISDIR(ma->ma_attr.la_mode)) {
1805                 rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_DEFAULT,
1806                                            0, handle);
1807                 if (rc == 0)
1808                         rc = mdo_declare_ref_add(env, p, handle);
1809         }
1810         if (rc)
1811                 GOTO(out, rc);
1812
1813         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_ACCESS,
1814                                    0, handle);
1815         if (rc)
1816                 GOTO(out, rc);
1817
1818         rc = mdd_declare_object_initialize(env, c, ma, handle);
1819         if (rc)
1820                 GOTO(out, rc);
1821
1822         rc = mdo_declare_index_insert(env, p, mdo2fid(c),
1823                                       name->ln_name, handle);
1824         if (rc)
1825                 GOTO(out, rc);
1826
1827         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV,
1828                                    0, handle);
1829         if (rc)
1830                 GOTO(out, rc);
1831
1832         if (S_ISLNK(ma->ma_attr.la_mode)) {
1833                 rc = dt_declare_record_write(env, mdd_object_child(c),
1834                                              strlen(spec->u.sp_symname), 0,
1835                                              handle);
1836                 if (rc)
1837                         GOTO(out, rc);
1838         }
1839         rc = mdo_declare_attr_set(env, p, &ma->ma_attr, handle);
1840         if (rc)
1841                 return rc;
1842
1843         rc = mdd_declare_changelog_store(env, mdd, name, handle);
1844         if (rc)
1845                 return rc;
1846
1847         rc = mdd_declare_lov_objid_update(env, mdd, handle);
1848
1849 out:
1850         return rc;
1851 }
1852
1853
1854 /*
1855  * Create object and insert it into namespace.
1856  */
1857 static int mdd_create(const struct lu_env *env,
1858                       struct md_object *pobj,
1859                       const struct lu_name *lname,
1860                       struct md_object *child,
1861                       struct md_op_spec *spec,
1862                       struct md_attr* ma)
1863 {
1864         struct mdd_thread_info *info = mdd_env_info(env);
1865         struct lu_attr         *la = &info->mti_la_for_fix;
1866         struct md_attr         *ma_acl = &info->mti_ma;
1867         struct mdd_object      *mdd_pobj = md2mdd_obj(pobj);
1868         struct mdd_object      *son = md2mdd_obj(child);
1869         struct mdd_device      *mdd = mdo2mdd(pobj);
1870         struct lu_attr         *attr = &ma->ma_attr;
1871         struct lov_mds_md      *lmm = NULL;
1872         struct thandle         *handle;
1873         struct dynlock_handle  *dlh;
1874         const char             *name = lname->ln_name;
1875         int rc, created = 0, initialized = 0, inserted = 0, lmm_size = 0;
1876         int got_def_acl = 0;
1877 #ifdef HAVE_QUOTA_SUPPORT
1878         struct obd_device *obd = mdd->mdd_obd_dev;
1879         struct obd_export *exp = md_quota(env)->mq_exp;
1880         struct mds_obd *mds = &obd->u.mds;
1881         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1882         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1883         int quota_opc = 0, block_count = 0;
1884         int inode_pending[MAXQUOTAS] = { 0, 0 };
1885         int block_pending[MAXQUOTAS] = { 0, 0 };
1886         int parent_pending[MAXQUOTAS] = { 0, 0 };
1887 #endif
1888         ENTRY;
1889
1890         /*
1891          * Two operations have to be performed:
1892          *
1893          *  - an allocation of a new object (->do_create()), and
1894          *
1895          *  - an insertion into a parent index (->dio_insert()).
1896          *
1897          * Due to locking, operation order is not important, when both are
1898          * successful, *but* error handling cases are quite different:
1899          *
1900          *  - if insertion is done first, and following object creation fails,
1901          *  insertion has to be rolled back, but this operation might fail
1902          *  also leaving us with dangling index entry.
1903          *
1904          *  - if creation is done first, is has to be undone if insertion
1905          *  fails, leaving us with leaked space, which is neither good, nor
1906          *  fatal.
1907          *
1908          * It seems that creation-first is simplest solution, but it is
1909          * sub-optimal in the frequent
1910          *
1911          *         $ mkdir foo
1912          *         $ mkdir foo
1913          *
1914          * case, because second mkdir is bound to create object, only to
1915          * destroy it immediately.
1916          *
1917          * To avoid this follow local file systems that do double lookup:
1918          *
1919          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
1920          *
1921          *     1. create            (mdd_object_create_internal())
1922          *
1923          *     2. insert            (__mdd_index_insert(), lookup again)
1924          */
1925
1926         /* Sanity checks before big job. */
1927         rc = mdd_create_sanity_check(env, pobj, lname, ma, spec);
1928         if (rc)
1929                 RETURN(rc);
1930
1931 #ifdef HAVE_QUOTA_SUPPORT
1932         if (mds->mds_quota) {
1933                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1934
1935                 rc = mdd_la_get(env, mdd_pobj, la_tmp, BYPASS_CAPA);
1936                 if (!rc) {
1937                         int same = 0;
1938
1939                         quota_opc = FSFILT_OP_CREATE;
1940                         mdd_quota_wrapper(&ma->ma_attr, qcids);
1941                         mdd_quota_wrapper(la_tmp, qpids);
1942                         /* get file quota for child */
1943                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1944                                         qcids, inode_pending, 1, NULL, 0, NULL,
1945                                         0);
1946                         switch (ma->ma_attr.la_mode & S_IFMT) {
1947                         case S_IFLNK:
1948                         case S_IFDIR:
1949                                 block_count = 2;
1950                                 break;
1951                         case S_IFREG:
1952                                 block_count = 1;
1953                                 break;
1954                         }
1955                         if (qcids[USRQUOTA] == qpids[USRQUOTA] &&
1956                             qcids[GRPQUOTA] == qpids[GRPQUOTA]) {
1957                                 block_count += 1;
1958                                 same = 1;
1959                         }
1960                         /* get block quota for child and parent */
1961                         if (block_count)
1962                                 lquota_chkquota(mds_quota_interface_ref, obd,
1963                                                 exp, qcids, block_pending,
1964                                                 block_count, NULL,
1965                                                 LQUOTA_FLAGS_BLK, NULL, 0);
1966                         if (!same)
1967                                 lquota_chkquota(mds_quota_interface_ref, obd,
1968                                                 exp, qpids, parent_pending, 1,
1969                                                 NULL, LQUOTA_FLAGS_BLK, NULL,
1970                                                 0);
1971                 }
1972         }
1973 #endif
1974
1975         /*
1976          * No RPC inside the transaction, so OST objects should be created at
1977          * first.
1978          */
1979         if (S_ISREG(attr->la_mode)) {
1980                 lmm_size = ma->ma_lmm_size;
1981                 rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
1982                                     spec, ma);
1983                 if (rc)
1984                         GOTO(out_pending, rc);
1985         }
1986
1987         if (!S_ISLNK(attr->la_mode)) {
1988                 ma_acl->ma_acl_size = sizeof info->mti_xattr_buf;
1989                 ma_acl->ma_acl = info->mti_xattr_buf;
1990                 ma_acl->ma_need = MA_ACL_DEF;
1991                 ma_acl->ma_valid = 0;
1992
1993                 mdd_read_lock(env, mdd_pobj, MOR_TGT_PARENT);
1994                 rc = mdd_def_acl_get(env, mdd_pobj, ma_acl);
1995                 mdd_read_unlock(env, mdd_pobj);
1996                 if (rc)
1997                         GOTO(out_free, rc);
1998                 else if (ma_acl->ma_valid & MA_ACL_DEF)
1999                         got_def_acl = 1;
2000         }
2001
2002         handle = mdd_trans_create(env, mdd);
2003         if (IS_ERR(handle))
2004                 GOTO(out_free, rc = PTR_ERR(handle));
2005
2006         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, ma,
2007                                 lmm_size, handle, spec);
2008         if (rc)
2009                 GOTO(out_stop, rc);
2010
2011         rc = mdd_trans_start(env, mdd, handle);
2012         if (rc)
2013                 GOTO(out_stop, rc);
2014
2015         dlh = mdd_pdo_write_lock(env, mdd_pobj, name, MOR_TGT_PARENT);
2016         if (dlh == NULL)
2017                 GOTO(out_trans, rc = -ENOMEM);
2018
2019         mdd_write_lock(env, son, MOR_TGT_CHILD);
2020         rc = mdd_object_create_internal(env, mdd_pobj, son, ma, handle, spec);
2021         if (rc) {
2022                 mdd_write_unlock(env, son);
2023                 GOTO(cleanup, rc);
2024         }
2025
2026         created = 1;
2027
2028 #ifdef CONFIG_FS_POSIX_ACL
2029         if (got_def_acl) {
2030                 struct lu_buf *acl_buf = &info->mti_buf;
2031                 acl_buf->lb_buf = ma_acl->ma_acl;
2032                 acl_buf->lb_len = ma_acl->ma_acl_size;
2033
2034                 rc = __mdd_acl_init(env, son, acl_buf, &attr->la_mode, handle);
2035                 if (rc) {
2036                         mdd_write_unlock(env, son);
2037                         GOTO(cleanup, rc);
2038                 } else {
2039                         ma->ma_attr.la_valid |= LA_MODE;
2040                 }
2041         }
2042 #endif
2043
2044         rc = mdd_object_initialize(env, mdo2fid(mdd_pobj), lname,
2045                                    son, ma, handle, spec);
2046         mdd_write_unlock(env, son);
2047         if (rc)
2048                 /*
2049                  * Object has no links, so it will be destroyed when last
2050                  * reference is released. (XXX not now.)
2051                  */
2052                 GOTO(cleanup, rc);
2053
2054         initialized = 1;
2055
2056         rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2057                                 name, S_ISDIR(attr->la_mode), handle,
2058                                 mdd_object_capa(env, mdd_pobj));
2059
2060         if (rc)
2061                 GOTO(cleanup, rc);
2062
2063         inserted = 1;
2064
2065         /* No need mdd_lsm_sanity_check here */
2066         rc = mdd_lov_set_md(env, mdd_pobj, son, lmm, lmm_size, handle, 0);
2067         if (rc) {
2068                 CERROR("error on stripe info copy %d \n", rc);
2069                 GOTO(cleanup, rc);
2070         }
2071         if (lmm && lmm_size > 0) {
2072                 /* Set Lov here, do not get lmm again later */
2073                 memcpy(ma->ma_lmm, lmm, lmm_size);
2074                 ma->ma_lmm_size = lmm_size;
2075                 ma->ma_valid |= MA_LOV;
2076         }
2077
2078         if (S_ISLNK(attr->la_mode)) {
2079                 struct md_ucred  *uc = md_ucred(env);
2080                 struct dt_object *dt = mdd_object_child(son);
2081                 const char *target_name = spec->u.sp_symname;
2082                 int sym_len = strlen(target_name);
2083                 const struct lu_buf *buf;
2084                 loff_t pos = 0;
2085
2086                 buf = mdd_buf_get_const(env, target_name, sym_len);
2087                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
2088                                                 mdd_object_capa(env, son),
2089                                                 uc->mu_cap &
2090                                                 CFS_CAP_SYS_RESOURCE_MASK);
2091
2092                 if (rc == sym_len)
2093                         rc = 0;
2094                 else
2095                         GOTO(cleanup, rc = -EFAULT);
2096         }
2097
2098         *la = ma->ma_attr;
2099         la->la_valid = LA_CTIME | LA_MTIME;
2100         rc = mdd_attr_check_set_internal_locked(env, mdd_pobj, la, handle, 0);
2101         if (rc)
2102                 GOTO(cleanup, rc);
2103
2104         /* Return attr back. */
2105         rc = mdd_attr_get_internal_locked(env, son, ma);
2106         EXIT;
2107 cleanup:
2108         if (rc && created) {
2109                 int rc2 = 0;
2110
2111                 if (inserted) {
2112                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2113                                                  S_ISDIR(attr->la_mode),
2114                                                  handle, BYPASS_CAPA);
2115                         if (rc2)
2116                                 CERROR("error can not cleanup destroy %d\n",
2117                                        rc2);
2118                 }
2119
2120                 if (rc2 == 0) {
2121                         mdd_write_lock(env, son, MOR_TGT_CHILD);
2122                         mdo_ref_del(env, son, handle);
2123                         if (initialized && S_ISDIR(attr->la_mode))
2124                                 mdo_ref_del(env, son, handle);
2125                         mdd_write_unlock(env, son);
2126                 }
2127         }
2128
2129         /* update lov_objid data, must be before transaction stop! */
2130         if (rc == 0)
2131                 mdd_lov_objid_update(mdd, lmm);
2132
2133         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
2134 out_trans:
2135         if (rc == 0)
2136                 rc = mdd_changelog_ns_store(env, mdd,
2137                             S_ISDIR(attr->la_mode) ? CL_MKDIR :
2138                             S_ISREG(attr->la_mode) ? CL_CREATE :
2139                             S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2140                             0, son, mdd_pobj, NULL, lname, handle);
2141 out_stop:
2142         mdd_trans_stop(env, mdd, rc, handle);
2143 out_free:
2144         /* finish lov_create stuff, free all temporary data */
2145         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
2146 out_pending:
2147 #ifdef HAVE_QUOTA_SUPPORT
2148         if (quota_opc) {
2149                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2150                                       inode_pending, 0);
2151                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2152                                       block_pending, 1);
2153                 lquota_pending_commit(mds_quota_interface_ref, obd, qpids,
2154                                       parent_pending, 1);
2155                 /* Trigger dqacq on the owner of child and parent. If failed,
2156                  * the next call for lquota_chkquota will process it. */
2157                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2158                               quota_opc);
2159         }
2160 #endif
2161         return rc;
2162 }
2163
2164 /*
2165  * Get locks on parents in proper order
2166  * RETURN: < 0 - error, rename_order if successful
2167  */
2168 enum rename_order {
2169         MDD_RN_SAME,
2170         MDD_RN_SRCTGT,
2171         MDD_RN_TGTSRC
2172 };
2173
2174 static int mdd_rename_order(const struct lu_env *env,
2175                             struct mdd_device *mdd,
2176                             struct mdd_object *src_pobj,
2177                             struct mdd_object *tgt_pobj)
2178 {
2179         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2180         int rc;
2181         ENTRY;
2182
2183         if (src_pobj == tgt_pobj)
2184                 RETURN(MDD_RN_SAME);
2185
2186         /* compared the parent child relationship of src_p&tgt_p */
2187         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2188                 rc = MDD_RN_SRCTGT;
2189         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2190                 rc = MDD_RN_TGTSRC;
2191         } else {
2192                 rc = mdd_is_parent(env, mdd, src_pobj, mdo2fid(tgt_pobj), NULL);
2193                 if (rc == -EREMOTE)
2194                         rc = 0;
2195
2196                 if (rc == 1)
2197                         rc = MDD_RN_TGTSRC;
2198                 else if (rc == 0)
2199                         rc = MDD_RN_SRCTGT;
2200         }
2201
2202         RETURN(rc);
2203 }
2204
2205 /* has not mdd_write{read}_lock on any obj yet. */
2206 static int mdd_rename_sanity_check(const struct lu_env *env,
2207                                    struct mdd_object *src_pobj,
2208                                    struct mdd_object *tgt_pobj,
2209                                    struct mdd_object *sobj,
2210                                    struct mdd_object *tobj,
2211                                    struct md_attr *ma)
2212 {
2213         int rc = 0;
2214         ENTRY;
2215
2216         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
2217                 RETURN(0);
2218
2219         /* XXX: when get here, sobj must NOT be NULL,
2220          * the other case has been processed in cml_rename
2221          * before mdd_rename and enable MDS_PERM_BYPASS. */
2222         LASSERT(sobj);
2223
2224         rc = mdd_may_delete(env, src_pobj, sobj, ma, 1, 0);
2225         if (rc)
2226                 RETURN(rc);
2227
2228         /* XXX: when get here, "tobj == NULL" means tobj must
2229          * NOT exist (neither on remote MDS, such case has been
2230          * processed in cml_rename before mdd_rename and enable
2231          * MDS_PERM_BYPASS).
2232          * So check may_create, but not check may_unlink. */
2233         if (!tobj)
2234                 rc = mdd_may_create(env, tgt_pobj, NULL,
2235                                     (src_pobj != tgt_pobj), 0);
2236         else
2237                 rc = mdd_may_delete(env, tgt_pobj, tobj, ma,
2238                                     (src_pobj != tgt_pobj), 1);
2239
2240         if (!rc && !tobj && (src_pobj != tgt_pobj) &&
2241             S_ISDIR(ma->ma_attr.la_mode))
2242                 rc = __mdd_may_link(env, tgt_pobj);
2243
2244         RETURN(rc);
2245 }
2246
2247 static int mdd_declare_rename(const struct lu_env *env,
2248                               struct mdd_device *mdd,
2249                               struct mdd_object *mdd_spobj,
2250                               struct mdd_object *mdd_tpobj,
2251                               struct mdd_object *mdd_sobj,
2252                               struct mdd_object *mdd_tobj,
2253                               const struct lu_name *sname,
2254                               const struct lu_name *tname,
2255                               struct md_attr *ma,
2256                               struct thandle *handle)
2257 {
2258         int rc;
2259
2260         LASSERT(mdd_spobj);
2261         LASSERT(mdd_tpobj);
2262         LASSERT(mdd_sobj);
2263
2264         /* name from source dir */
2265         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2266         if (rc)
2267                 return rc;
2268
2269         /* .. from source child */
2270         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2271                 /* source child can be directory,
2272                  * counted by source dir's nlink */
2273                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2274                 if (rc)
2275                         return rc;
2276
2277                 rc = mdo_declare_index_delete(env, mdd_sobj, dotdot, handle);
2278                 if (rc)
2279                         return rc;
2280
2281                 rc = mdo_declare_index_insert(env, mdd_sobj, mdo2fid(mdd_tpobj),
2282                                               dotdot, handle);
2283                 if (rc)
2284                         return rc;
2285
2286                 /* new target child can be directory,
2287                  * counted by target dir's nlink */
2288                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2289                 if (rc)
2290                         return rc;
2291
2292         }
2293
2294         rc = mdo_declare_attr_set(env, mdd_spobj, NULL, handle);
2295         if (rc)
2296                 return rc;
2297
2298         rc = mdo_declare_attr_set(env, mdd_sobj, NULL, handle);
2299         if (rc)
2300                 return rc;
2301         mdd_declare_links_add(env, mdd_sobj, handle);
2302         if (rc)
2303                 return rc;
2304
2305         rc = mdo_declare_attr_set(env, mdd_tpobj, NULL, handle);
2306         if (rc)
2307                 return rc;
2308
2309         /* new name */
2310         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2311                         tname->ln_name, handle);
2312         if (rc)
2313                 return rc;
2314
2315         /* name from target dir (old name), we declare it unconditionally
2316          * as mdd_rename() calls delete unconditionally as well. so just
2317          * to balance declarations vs calls to change ... */
2318         rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name, handle);
2319         if (rc)
2320                 return rc;
2321
2322         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2323                 /* delete target child in target parent directory */
2324                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2325                 if (rc)
2326                         return rc;
2327
2328                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2329                         /* target child can be directory,
2330                          * delete "." reference in target child directory */
2331                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2332                         if (rc)
2333                                 return rc;
2334
2335                         /* delete ".." reference in target parent directory */
2336                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2337                         if (rc)
2338                                 return rc;
2339                 }
2340
2341                 rc = mdo_declare_attr_set(env, mdd_tobj, NULL, handle);
2342                 if (rc)
2343                         return rc;
2344
2345                 mdd_declare_links_add(env, mdd_tobj, handle);
2346                 if (rc)
2347                         return rc;
2348
2349                 rc = mdd_declare_finish_unlink(env, mdd_tobj, ma, handle);
2350                 if (rc)
2351                         return rc;
2352         }
2353
2354         rc = mdd_declare_changelog_store(env, mdd, tname, handle);
2355         if (rc)
2356                 return rc;
2357
2358         rc = mdd_declare_changelog_store(env, mdd, sname, handle);
2359         if (rc)
2360                 return rc;
2361
2362         return rc;
2363 }
2364
2365 /* src object can be remote that is why we use only fid and type of object */
2366 static int mdd_rename(const struct lu_env *env,
2367                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2368                       const struct lu_fid *lf, const struct lu_name *lsname,
2369                       struct md_object *tobj, const struct lu_name *ltname,
2370                       struct md_attr *ma)
2371 {
2372         const char *sname = lsname->ln_name;
2373         const char *tname = ltname->ln_name;
2374         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2375         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2376         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2377         struct mdd_device *mdd = mdo2mdd(src_pobj);
2378         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2379         struct mdd_object *mdd_tobj = NULL;
2380         struct dynlock_handle *sdlh, *tdlh;
2381         struct thandle *handle;
2382         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2383         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2384         int is_dir;
2385         int rc, rc2;
2386
2387 #ifdef HAVE_QUOTA_SUPPORT
2388         struct obd_device *obd = mdd->mdd_obd_dev;
2389         struct obd_export *exp = md_quota(env)->mq_exp;
2390         struct mds_obd *mds = &obd->u.mds;
2391         unsigned int qspids[MAXQUOTAS] = { 0, 0 };
2392         unsigned int qtcids[MAXQUOTAS] = { 0, 0 };
2393         unsigned int qtpids[MAXQUOTAS] = { 0, 0 };
2394         int quota_copc = 0, quota_popc = 0;
2395         int rec_pending[MAXQUOTAS] = { 0, 0 };
2396 #endif
2397         ENTRY;
2398
2399         LASSERT(ma->ma_attr.la_mode & S_IFMT);
2400         is_dir = S_ISDIR(ma->ma_attr.la_mode);
2401
2402         if (tobj)
2403                 mdd_tobj = md2mdd_obj(tobj);
2404
2405 #ifdef HAVE_QUOTA_SUPPORT
2406         if (mds->mds_quota) {
2407                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
2408
2409                 rc = mdd_la_get(env, mdd_spobj, la_tmp, BYPASS_CAPA);
2410                 if (!rc) {
2411                         mdd_quota_wrapper(la_tmp, qspids);
2412                         if (!tobj) {
2413                                 rc = mdd_la_get(env, mdd_tpobj, la_tmp,
2414                                                 BYPASS_CAPA);
2415                                 if (!rc) {
2416                                         void *data = NULL;
2417                                         mdd_data_get(env, mdd_tpobj, &data);
2418                                         quota_popc = FSFILT_OP_LINK;
2419                                         mdd_quota_wrapper(la_tmp, qtpids);
2420                                         /* get block quota for target parent */
2421                                         lquota_chkquota(mds_quota_interface_ref,
2422                                                         obd, exp, qtpids,
2423                                                         rec_pending, 1, NULL,
2424                                                         LQUOTA_FLAGS_BLK,
2425                                                         data, 1);
2426                                 }
2427                         }
2428                 }
2429         }
2430 #endif
2431         mdd_sobj = mdd_object_find(env, mdd, lf);
2432
2433         handle = mdd_trans_create(env, mdd);
2434         if (IS_ERR(handle))
2435                 GOTO(out_pending, rc = PTR_ERR(handle));
2436
2437         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
2438                                 mdd_tobj, lsname, ltname, ma, handle);
2439         if (rc)
2440                 GOTO(stop, rc);
2441
2442         rc = mdd_trans_start(env, mdd, handle);
2443         if (rc)
2444                 GOTO(stop, rc);
2445
2446         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2447         rc = mdd_rename_order(env, mdd, mdd_spobj, mdd_tpobj);
2448         if (rc < 0)
2449                 GOTO(cleanup_unlocked, rc);
2450
2451         /* Get locks in determined order */
2452         if (rc == MDD_RN_SAME) {
2453                 sdlh = mdd_pdo_write_lock(env, mdd_spobj,
2454                                           sname, MOR_SRC_PARENT);
2455                 /* check hashes to determine do we need one lock or two */
2456                 if (mdd_name2hash(sname) != mdd_name2hash(tname))
2457                         tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,
2458                                 MOR_TGT_PARENT);
2459                 else
2460                         tdlh = sdlh;
2461         } else if (rc == MDD_RN_SRCTGT) {
2462                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_SRC_PARENT);
2463                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_TGT_PARENT);
2464         } else {
2465                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_SRC_PARENT);
2466                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_TGT_PARENT);
2467         }
2468         if (sdlh == NULL || tdlh == NULL)
2469                 GOTO(cleanup, rc = -ENOMEM);
2470
2471         rc = mdd_rename_sanity_check(env, mdd_spobj, mdd_tpobj,
2472                                      mdd_sobj, mdd_tobj, ma);
2473         if (rc)
2474                 GOTO(cleanup, rc);
2475
2476         /* Remove source name from source directory */
2477         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle,
2478                                 mdd_object_capa(env, mdd_spobj));
2479         if (rc)
2480                 GOTO(cleanup, rc);
2481
2482         /* "mv dir1 dir2" needs "dir1/.." link update */
2483         if (is_dir && mdd_sobj && !lu_fid_eq(spobj_fid, tpobj_fid)) {
2484                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2485                                         mdd_object_capa(env, mdd_sobj));
2486                 if (rc)
2487                         GOTO(fixup_spobj2, rc);
2488
2489                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, dotdot,
2490                                       handle, mdd_object_capa(env, mdd_sobj));
2491                 if (rc)
2492                         GOTO(fixup_spobj, rc);
2493         }
2494
2495         /* Remove target name from target directory
2496          * Here tobj can be remote one, so we do index_delete unconditionally
2497          * and -ENOENT is allowed.
2498          */
2499         rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2500                                 mdd_object_capa(env, mdd_tpobj));
2501         if (rc != 0) {
2502                 if (mdd_tobj) {
2503                         /* tname might been renamed to something else */
2504                         GOTO(fixup_spobj, rc);
2505                 }
2506                 if (rc != -ENOENT)
2507                         GOTO(fixup_spobj, rc);
2508         }
2509
2510         /* Insert new fid with target name into target dir */
2511         rc = __mdd_index_insert(env, mdd_tpobj, lf, tname, is_dir, handle,
2512                                 mdd_object_capa(env, mdd_tpobj));
2513         if (rc)
2514                 GOTO(fixup_tpobj, rc);
2515
2516         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2517         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2518
2519         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
2520         if (mdd_sobj) {
2521                 la->la_valid = LA_CTIME;
2522                 rc = mdd_attr_check_set_internal_locked(env, mdd_sobj, la,
2523                                                         handle, 0);
2524                 if (rc)
2525                         GOTO(fixup_tpobj, rc);
2526         }
2527
2528         /* Remove old target object
2529          * For tobj is remote case cmm layer has processed
2530          * and set tobj to NULL then. So when tobj is NOT NULL,
2531          * it must be local one.
2532          */
2533         if (tobj && mdd_object_exists(mdd_tobj)) {
2534                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
2535                 if (mdd_is_dead_obj(mdd_tobj)) {
2536                         mdd_write_unlock(env, mdd_tobj);
2537                         /* shld not be dead, something is wrong */
2538                         CERROR("tobj is dead, something is wrong\n");
2539                         rc = -EINVAL;
2540                         goto cleanup;
2541                 }
2542                 mdo_ref_del(env, mdd_tobj, handle);
2543
2544                 /* Remove dot reference. */
2545                 if (is_dir)
2546                         mdo_ref_del(env, mdd_tobj, handle);
2547
2548                 la->la_valid = LA_CTIME;
2549                 rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
2550                 if (rc)
2551                         GOTO(fixup_tpobj, rc);
2552
2553                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
2554                 mdd_write_unlock(env, mdd_tobj);
2555                 if (rc)
2556                         GOTO(fixup_tpobj, rc);
2557
2558 #ifdef HAVE_QUOTA_SUPPORT
2559                 if (mds->mds_quota && ma->ma_valid & MA_INODE &&
2560                     ma->ma_attr.la_nlink == 0 && mdd_tobj->mod_count == 0) {
2561                         quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
2562                         mdd_quota_wrapper(&ma->ma_attr, qtcids);
2563                 }
2564 #endif
2565         }
2566
2567         la->la_valid = LA_CTIME | LA_MTIME;
2568         rc = mdd_attr_check_set_internal_locked(env, mdd_spobj, la, handle, 0);
2569         if (rc)
2570                 GOTO(fixup_tpobj, rc);
2571
2572         if (mdd_spobj != mdd_tpobj) {
2573                 la->la_valid = LA_CTIME | LA_MTIME;
2574                 rc = mdd_attr_check_set_internal_locked(env, mdd_tpobj, la,
2575                                                   handle, 0);
2576         }
2577
2578         if (rc == 0 && mdd_sobj) {
2579                 mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
2580                 rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
2581                                       mdo2fid(mdd_tpobj), ltname, handle);
2582                 if (rc == -ENOENT)
2583                         /* Old files might not have EA entry */
2584                         mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
2585                                       lsname, handle, 0);
2586                 mdd_write_unlock(env, mdd_sobj);
2587                 /* We don't fail the transaction if the link ea can't be
2588                    updated -- fid2path will use alternate lookup method. */
2589                 rc = 0;
2590         }
2591
2592         EXIT;
2593
2594 fixup_tpobj:
2595         if (rc) {
2596                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2597                                          BYPASS_CAPA);
2598                 if (rc2)
2599                         CWARN("tp obj fix error %d\n",rc2);
2600
2601                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
2602                     !mdd_is_dead_obj(mdd_tobj)) {
2603                         rc2 = __mdd_index_insert(env, mdd_tpobj,
2604                                          mdo2fid(mdd_tobj), tname,
2605                                          is_dir, handle,
2606                                          BYPASS_CAPA);
2607
2608                         if (rc2)
2609                                 CWARN("tp obj fix error %d\n",rc2);
2610                 }
2611         }
2612
2613 fixup_spobj:
2614         if (rc && is_dir && mdd_sobj) {
2615                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2616                                               BYPASS_CAPA);
2617
2618                 if (rc2)
2619                         CWARN("sp obj dotdot delete error %d\n",rc2);
2620
2621
2622                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid,
2623                                               dotdot, handle, BYPASS_CAPA);
2624                 if (rc2)
2625                         CWARN("sp obj dotdot insert error %d\n",rc2);
2626         }
2627
2628 fixup_spobj2:
2629         if (rc) {
2630                 rc2 = __mdd_index_insert(env, mdd_spobj,
2631                                          lf, sname, is_dir, handle, BYPASS_CAPA);
2632                 if (rc2)
2633                         CWARN("sp obj fix error %d\n",rc2);
2634         }
2635 cleanup:
2636         if (likely(tdlh) && sdlh != tdlh)
2637                 mdd_pdo_write_unlock(env, mdd_tpobj, tdlh);
2638         if (likely(sdlh))
2639                 mdd_pdo_write_unlock(env, mdd_spobj, sdlh);
2640 cleanup_unlocked:
2641         if (rc == 0)
2642                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, 0, mdd_tobj,
2643                                             mdd_spobj, lf, lsname, handle);
2644         if (rc == 0) {
2645                 struct lu_fid zero_fid;
2646                 fid_zero(&zero_fid);
2647                 /* If the rename target exist, The CL_EXT record should save
2648                  * the target fid as tfid, otherwise, use zero fid. LU-543 */
2649                 rc = mdd_changelog_ns_store(env, mdd, CL_EXT, 0, mdd_tobj,
2650                                             mdd_tpobj,
2651                                             mdd_tobj ? NULL : &zero_fid,
2652                                             ltname, handle);
2653         }
2654
2655 stop:
2656         mdd_trans_stop(env, mdd, rc, handle);
2657         if (mdd_sobj)
2658                 mdd_object_put(env, mdd_sobj);
2659 out_pending:
2660 #ifdef HAVE_QUOTA_SUPPORT
2661         if (mds->mds_quota) {
2662                 if (quota_popc)
2663                         lquota_pending_commit(mds_quota_interface_ref, obd,
2664                                               qtpids, rec_pending, 1);
2665
2666                 if (quota_copc) {
2667                         /* Trigger dqrel on the source owner of parent.
2668                          * If failed, the next call for lquota_chkquota will
2669                          * process it. */
2670                         lquota_adjust(mds_quota_interface_ref, obd, 0, qspids, rc,
2671                                       FSFILT_OP_UNLINK_PARTIAL_PARENT);
2672
2673                         /* Trigger dqrel on the target owner of child.
2674                          * If failed, the next call for lquota_chkquota
2675                          * will process it. */
2676                         lquota_adjust(mds_quota_interface_ref, obd, qtcids,
2677                                       qtpids, rc, quota_copc);
2678                 }
2679         }
2680 #endif
2681         return rc;
2682 }
2683
2684 /** enable/disable storing of hardlink info */
2685 int mdd_linkea_enable = 1;
2686 CFS_MODULE_PARM(mdd_linkea_enable, "d", int, 0644,
2687                 "record hardlink info in EAs");
2688
2689 /** Read the link EA into a temp buffer.
2690  * Uses the name_buf since it is generally large.
2691  * \retval IS_ERR err
2692  * \retval ptr to \a lu_buf (always \a mti_big_buf)
2693  */
2694 struct lu_buf *mdd_links_get(const struct lu_env *env,
2695                              struct mdd_object *mdd_obj)
2696 {
2697         struct lu_buf *buf;
2698         struct lustre_capa *capa;
2699         struct link_ea_header *leh;
2700         int rc;
2701
2702         /* First try a small buf */
2703         buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2704         if (buf->lb_buf == NULL)
2705                 return ERR_PTR(-ENOMEM);
2706
2707         capa = mdd_object_capa(env, mdd_obj);
2708         rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2709         if (rc == -ERANGE) {
2710                 /* Buf was too small, figure out what we need. */
2711                 mdd_buf_put(buf);
2712                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2713                 if (rc < 0)
2714                         return ERR_PTR(rc);
2715                 buf = mdd_buf_alloc(env, rc);
2716                 if (buf->lb_buf == NULL)
2717                         return ERR_PTR(-ENOMEM);
2718                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2719         }
2720         if (rc < 0)
2721                 return ERR_PTR(rc);
2722
2723         leh = buf->lb_buf;
2724         if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
2725                 leh->leh_magic = LINK_EA_MAGIC;
2726                 leh->leh_reccount = __swab32(leh->leh_reccount);
2727                 leh->leh_len = __swab64(leh->leh_len);
2728                 /* entries are swabbed by mdd_lee_unpack */
2729         }
2730         if (leh->leh_magic != LINK_EA_MAGIC)
2731                 return ERR_PTR(-EINVAL);
2732         if (leh->leh_reccount == 0)
2733                 return ERR_PTR(-ENODATA);
2734
2735         return buf;
2736 }
2737
2738 /** Pack a link_ea_entry.
2739  * All elements are stored as chars to avoid alignment issues.
2740  * Numbers are always big-endian
2741  * \retval record length
2742  */
2743 static int mdd_lee_pack(struct link_ea_entry *lee, const struct lu_name *lname,
2744                         const struct lu_fid *pfid)
2745 {
2746         struct lu_fid   tmpfid;
2747         int             reclen;
2748
2749         fid_cpu_to_be(&tmpfid, pfid);
2750         memcpy(&lee->lee_parent_fid, &tmpfid, sizeof(tmpfid));
2751         memcpy(lee->lee_name, lname->ln_name, lname->ln_namelen);
2752         reclen = sizeof(struct link_ea_entry) + lname->ln_namelen;
2753
2754         lee->lee_reclen[0] = (reclen >> 8) & 0xff;
2755         lee->lee_reclen[1] = reclen & 0xff;
2756         return reclen;
2757 }
2758
2759 void mdd_lee_unpack(const struct link_ea_entry *lee, int *reclen,
2760                     struct lu_name *lname, struct lu_fid *pfid)
2761 {
2762         *reclen = (lee->lee_reclen[0] << 8) | lee->lee_reclen[1];
2763         memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
2764         fid_be_to_cpu(pfid, pfid);
2765         lname->ln_name = lee->lee_name;
2766         lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
2767 }
2768
2769 /** Add a record to the end of link ea buf */
2770 static int __mdd_links_add(const struct lu_env *env, struct lu_buf *buf,
2771                            const struct lu_fid *pfid,
2772                            const struct lu_name *lname)
2773 {
2774         struct link_ea_header *leh;
2775         struct link_ea_entry *lee;
2776         int reclen;
2777
2778         if (lname == NULL || pfid == NULL)
2779                 return -EINVAL;
2780
2781         /* Make sure our buf is big enough for the new one */
2782         leh = buf->lb_buf;
2783         reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
2784         if (leh->leh_len + reclen > buf->lb_len) {
2785                 if (mdd_buf_grow(env, leh->leh_len + reclen) < 0)
2786                         return -ENOMEM;
2787         }
2788
2789         leh = buf->lb_buf;
2790         lee = buf->lb_buf + leh->leh_len;
2791         reclen = mdd_lee_pack(lee, lname, pfid);
2792         leh->leh_len += reclen;
2793         leh->leh_reccount++;
2794         return 0;
2795 }
2796
2797 static int mdd_declare_links_add(const struct lu_env *env,
2798                                  struct mdd_object *mdd_obj,
2799                                  struct thandle *handle)
2800 {
2801         int rc;
2802
2803         /* XXX: max size? */
2804         rc = mdo_declare_xattr_set(env, mdd_obj,
2805                              mdd_buf_get_const(env, NULL, 4096),
2806                              XATTR_NAME_LINK, 0, handle);
2807
2808         return rc;
2809 }
2810
2811 /* For pathologic linkers, we don't want to spend lots of time scanning the
2812  * link ea.  Limit ourseleves to something reasonable; links not in the EA
2813  * can be looked up via (slower) parent lookup.
2814  */
2815 #define LINKEA_MAX_COUNT 128
2816
2817 static int mdd_links_add(const struct lu_env *env,
2818                          struct mdd_object *mdd_obj,
2819                          const struct lu_fid *pfid,
2820                          const struct lu_name *lname,
2821                          struct thandle *handle, int first)
2822 {
2823         struct lu_buf *buf;
2824         struct link_ea_header *leh;
2825         int rc;
2826         ENTRY;
2827
2828         if (!mdd_linkea_enable)
2829                 RETURN(0);
2830
2831         buf = first ? ERR_PTR(-ENODATA) : mdd_links_get(env, mdd_obj);
2832         if (IS_ERR(buf)) {
2833                 rc = PTR_ERR(buf);
2834                 if (rc != -ENODATA) {
2835                         CERROR("link_ea read failed %d "DFID"\n", rc,
2836                                PFID(mdd_object_fid(mdd_obj)));
2837                         RETURN (rc);
2838                 }
2839                 /* empty EA; start one */
2840                 buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2841                 if (buf->lb_buf == NULL)
2842                         RETURN(-ENOMEM);
2843                 leh = buf->lb_buf;
2844                 leh->leh_magic = LINK_EA_MAGIC;
2845                 leh->leh_len = sizeof(struct link_ea_header);
2846                 leh->leh_reccount = 0;
2847         }
2848
2849         leh = buf->lb_buf;
2850         if (leh->leh_reccount > LINKEA_MAX_COUNT)
2851                 RETURN(-EOVERFLOW);
2852
2853         rc = __mdd_links_add(env, buf, pfid, lname);
2854         if (rc)
2855                 RETURN(rc);
2856
2857         leh = buf->lb_buf;
2858         rc = __mdd_xattr_set(env, mdd_obj,
2859                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2860                              XATTR_NAME_LINK, 0, handle);
2861         if (rc) {
2862                 if (rc == -ENOSPC)
2863                         CDEBUG(D_INODE, "link_ea add failed %d "DFID"\n", rc,
2864                                PFID(mdd_object_fid(mdd_obj)));
2865                 else
2866                         CERROR("link_ea add failed %d "DFID"\n", rc,
2867                                PFID(mdd_object_fid(mdd_obj)));
2868         }
2869
2870         if (buf->lb_len > OBD_ALLOC_BIG)
2871                 /* if we vmalloced a large buffer drop it */
2872                 mdd_buf_put(buf);
2873
2874         RETURN (rc);
2875 }
2876
2877 static int mdd_links_rename(const struct lu_env *env,
2878                             struct mdd_object *mdd_obj,
2879                             const struct lu_fid *oldpfid,
2880                             const struct lu_name *oldlname,
2881                             const struct lu_fid *newpfid,
2882                             const struct lu_name *newlname,
2883                             struct thandle *handle)
2884 {
2885         struct lu_buf  *buf;
2886         struct link_ea_header *leh;
2887         struct link_ea_entry  *lee;
2888         struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
2889         struct lu_fid  *tmpfid = &mdd_env_info(env)->mti_fid;
2890         int reclen = 0;
2891         int count;
2892         int rc, rc2 = 0;
2893         ENTRY;
2894
2895         if (!mdd_linkea_enable)
2896                 RETURN(0);
2897
2898         if (mdd_obj->mod_flags & DEAD_OBJ)
2899                 /* No more links, don't bother */
2900                 RETURN(0);
2901
2902         buf = mdd_links_get(env, mdd_obj);
2903         if (IS_ERR(buf)) {
2904                 rc = PTR_ERR(buf);
2905                 if (rc == -ENODATA)
2906                         CDEBUG(D_INODE, "link_ea read failed %d "DFID"\n",
2907                                rc, PFID(mdd_object_fid(mdd_obj)));
2908                 else
2909                         CERROR("link_ea read failed %d "DFID"\n",
2910                                rc, PFID(mdd_object_fid(mdd_obj)));
2911                 RETURN(rc);
2912         }
2913         leh = buf->lb_buf;
2914         lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
2915
2916         /* Find the old record */
2917         for(count = 0; count < leh->leh_reccount; count++) {
2918                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
2919                 if (tmpname->ln_namelen == oldlname->ln_namelen &&
2920                     lu_fid_eq(tmpfid, oldpfid) &&
2921                     (strncmp(tmpname->ln_name, oldlname->ln_name,
2922                              tmpname->ln_namelen) == 0))
2923                         break;
2924                 lee = (struct link_ea_entry *)((char *)lee + reclen);
2925         }
2926         if ((count + 1) > leh->leh_reccount) {
2927                 CDEBUG(D_INODE, "Old link_ea name '%.*s' not found\n",
2928                        oldlname->ln_namelen, oldlname->ln_name);
2929                 GOTO(out, rc = -ENOENT);
2930         }
2931
2932         /* Remove the old record */
2933         leh->leh_reccount--;
2934         leh->leh_len -= reclen;
2935         memmove(lee, (char *)lee + reclen, (char *)leh + leh->leh_len -
2936                 (char *)lee);
2937
2938         /* If renaming, add the new record */
2939         if (newpfid != NULL) {
2940                 /* if the add fails, we still delete the out-of-date old link */
2941                 rc2 = __mdd_links_add(env, buf, newpfid, newlname);
2942                 leh = buf->lb_buf;
2943         }
2944
2945         rc = __mdd_xattr_set(env, mdd_obj,
2946                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2947                              XATTR_NAME_LINK, 0, handle);
2948
2949 out:
2950         if (rc == 0)
2951                 rc = rc2;
2952         if (rc)
2953                 CDEBUG(D_INODE, "link_ea mv/unlink '%.*s' failed %d "DFID"\n",
2954                        oldlname->ln_namelen, oldlname->ln_name, rc,
2955                        PFID(mdd_object_fid(mdd_obj)));
2956
2957         if (buf->lb_len > OBD_ALLOC_BIG)
2958                 /* if we vmalloced a large buffer drop it */
2959                 mdd_buf_put(buf);
2960
2961         RETURN (rc);
2962 }
2963
2964 const struct md_dir_operations mdd_dir_ops = {
2965         .mdo_is_subdir     = mdd_is_subdir,
2966         .mdo_lookup        = mdd_lookup,
2967         .mdo_create        = mdd_create,
2968         .mdo_rename        = mdd_rename,
2969         .mdo_link          = mdd_link,
2970         .mdo_unlink        = mdd_unlink,
2971         .mdo_lum_lmm_cmp   = mdd_lum_lmm_cmp,
2972         .mdo_name_insert   = mdd_name_insert,
2973         .mdo_name_remove   = mdd_name_remove,
2974         .mdo_rename_tgt    = mdd_rename_tgt,
2975         .mdo_create_data   = mdd_create_data,
2976 };