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