lustre/mdd/mdd_dir.c

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