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,
1082                                        ma->ma_lmm, ma->ma_lmm_size,
1083                                        ma->ma_cookie, 1);
1084 #else
1085 #warning "please remove this after 2.4 (LOD/OSP)."
1086 #endif
1087
1088 #ifdef HAVE_QUOTA_SUPPORT
1089         if (quota_opc)
1090                 /* Trigger dqrel on the owner of child and parent. If failed,
1091                  * the next call for lquota_chkquota will process it. */
1092                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
1093                               quota_opc);
1094 #endif
1095         return rc;
1096 }
1097
1098 /* has not lock on pobj yet */
1099 static int mdd_ni_sanity_check(const struct lu_env *env,
1100                                struct md_object *pobj,
1101                                const struct md_attr *ma)
1102 {
1103         struct mdd_object *obj = md2mdd_obj(pobj);
1104         int rc;
1105         ENTRY;
1106
1107         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
1108                 RETURN(0);
1109
1110         rc = mdd_may_create(env, obj, NULL, 1, S_ISDIR(ma->ma_attr.la_mode));
1111
1112         RETURN(rc);
1113 }
1114
1115 /*
1116  * Partial operation.
1117  */
1118 static int mdd_name_insert(const struct lu_env *env,
1119                            struct md_object *pobj,
1120                            const struct lu_name *lname,
1121                            const struct lu_fid *fid,
1122                            const struct md_attr *ma)
1123 {
1124         const char *name = lname->ln_name;
1125         struct lu_attr   *la = &mdd_env_info(env)->mti_la_for_fix;
1126         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1127         struct dynlock_handle *dlh;
1128         struct thandle *handle;
1129         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1130 #ifdef HAVE_QUOTA_SUPPORT
1131         struct mdd_device *mdd = mdo2mdd(pobj);
1132         struct md_ucred *uc = md_ucred(env);
1133         struct obd_device *obd = mdd->mdd_obd_dev;
1134         struct obd_export *exp = md_quota(env)->mq_exp;
1135         struct mds_obd *mds = &obd->u.mds;
1136         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1137         int quota_opc = 0, rec_pending[MAXQUOTAS] = { 0, 0 };
1138         cfs_cap_t save = uc->mu_cap;
1139 #endif
1140         int rc;
1141         ENTRY;
1142
1143         /* XXX: this code won't be used ever:
1144          * DNE uses slightly different approach */
1145         LBUG();
1146
1147 #ifdef HAVE_QUOTA_SUPPORT
1148         if (mds->mds_quota) {
1149                 if (!(ma->ma_attr_flags & MDS_QUOTA_IGNORE)) {
1150                         struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1151
1152                         rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1153                         if (!rc) {
1154                                 void *data = NULL;
1155                                 mdd_data_get(env, mdd_obj, &data);
1156                                 quota_opc = FSFILT_OP_LINK;
1157                                 mdd_quota_wrapper(la_tmp, qids);
1158                                 /* get block quota for parent */
1159                                 lquota_chkquota(mds_quota_interface_ref, obd,
1160                                                 exp, qids, rec_pending, 1, NULL,
1161                                                 LQUOTA_FLAGS_BLK, data, 1);
1162                         }
1163                 } else {
1164                         uc->mu_cap |= CFS_CAP_SYS_RESOURCE_MASK;
1165                 }
1166         }
1167 #endif
1168         handle = mdd_trans_create(env, mdo2mdd(pobj));
1169         if (IS_ERR(handle))
1170                 GOTO(out_pending, rc = PTR_ERR(handle));
1171
1172         rc = mdd_trans_start(env, mdo2mdd(pobj), handle);
1173
1174         dlh = mdd_pdo_write_lock(env, mdd_obj, name, MOR_TGT_PARENT);
1175         if (dlh == NULL)
1176                 GOTO(out_trans, rc = -ENOMEM);
1177
1178         rc = mdd_ni_sanity_check(env, pobj, ma);
1179         if (rc)
1180                 GOTO(out_unlock, rc);
1181
1182         rc = __mdd_index_insert(env, mdd_obj, fid, name, is_dir,
1183                                 handle, BYPASS_CAPA);
1184         if (rc)
1185                 GOTO(out_unlock, rc);
1186
1187         /*
1188          * For some case, no need update obj's ctime (LA_CTIME is not set),
1189          * e.g. split_dir.
1190          * For other cases, update obj's ctime (LA_CTIME is set),
1191          * e.g. cmr_link.
1192          */
1193         if (ma->ma_attr.la_valid & LA_CTIME) {
1194                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1195                 la->la_valid = LA_CTIME | LA_MTIME;
1196                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la,
1197                                                         handle, 0);
1198         }
1199         EXIT;
1200 out_unlock:
1201         mdd_pdo_write_unlock(env, mdd_obj, dlh);
1202 out_trans:
1203         mdd_trans_stop(env, mdo2mdd(pobj), rc, handle);
1204 out_pending:
1205 #ifdef HAVE_QUOTA_SUPPORT
1206         if (mds->mds_quota) {
1207                 if (quota_opc) {
1208                         lquota_pending_commit(mds_quota_interface_ref,
1209                                               obd, qids, rec_pending, 1);
1210                         /* Trigger dqacq for the parent owner. If failed,
1211                          * the next call for lquota_chkquota will process it*/
1212                         lquota_adjust(mds_quota_interface_ref, obd, 0, qids,
1213                                       rc, quota_opc);
1214                 } else {
1215                         uc->mu_cap = save;
1216                 }
1217         }
1218 #endif
1219         return rc;
1220 }
1221
1222 /* has not lock on pobj yet */
1223 static int mdd_nr_sanity_check(const struct lu_env *env,
1224                                struct md_object *pobj,
1225                                const struct md_attr *ma)
1226 {
1227         struct mdd_object *obj = md2mdd_obj(pobj);
1228         int rc;
1229         ENTRY;
1230
1231         if (ma->ma_attr_flags & MDS_PERM_BYPASS)
1232                 RETURN(0);
1233
1234         rc = mdd_may_unlink(env, obj, ma);
1235
1236         RETURN(rc);
1237 }
1238
1239 /*
1240  * Partial operation.
1241  */
1242 static int mdd_name_remove(const struct lu_env *env,
1243                            struct md_object *pobj,
1244                            const struct lu_name *lname,
1245                            const struct md_attr *ma)
1246 {
1247         const char *name = lname->ln_name;
1248         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1249         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1250         struct mdd_device *mdd = mdo2mdd(pobj);
1251         struct dynlock_handle *dlh;
1252         struct thandle *handle;
1253         int is_dir = S_ISDIR(ma->ma_attr.la_mode);
1254 #ifdef HAVE_QUOTA_SUPPORT
1255         struct obd_device *obd = mdd->mdd_obd_dev;
1256         struct mds_obd *mds = &obd->u.mds;
1257         unsigned int qids[MAXQUOTAS] = { 0, 0 };
1258         int quota_opc = 0;
1259 #endif
1260         int rc;
1261         ENTRY;
1262
1263         /* XXX: this code won't be used ever:
1264          * DNE uses slightly different approach */
1265         LBUG();
1266
1267 #ifdef HAVE_QUOTA_SUPPORT
1268         if (mds->mds_quota) {
1269                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1270
1271                 rc = mdd_la_get(env, mdd_obj, la_tmp, BYPASS_CAPA);
1272                 if (!rc) {
1273                         quota_opc = FSFILT_OP_UNLINK_PARTIAL_PARENT;
1274                         mdd_quota_wrapper(la_tmp, qids);
1275                 }
1276         }
1277 #endif
1278         handle = mdd_trans_create(env, mdd);
1279         if (IS_ERR(handle))
1280                 GOTO(out_pending, rc = PTR_ERR(handle));
1281
1282         rc = mdd_trans_start(env, mdd, handle);
1283
1284         dlh = mdd_pdo_write_lock(env, mdd_obj, name, MOR_TGT_PARENT);
1285         if (dlh == NULL)
1286                 GOTO(out_trans, rc = -ENOMEM);
1287
1288         rc = mdd_nr_sanity_check(env, pobj, ma);
1289         if (rc)
1290                 GOTO(out_unlock, rc);
1291
1292         rc = __mdd_index_delete(env, mdd_obj, name, is_dir,
1293                                 handle, BYPASS_CAPA);
1294         if (rc)
1295                 GOTO(out_unlock, rc);
1296
1297         /*
1298          * For some case, no need update obj's ctime (LA_CTIME is not set),
1299          * e.g. split_dir.
1300          * For other cases, update obj's ctime (LA_CTIME is set),
1301          * e.g. cmr_unlink.
1302          */
1303         if (ma->ma_attr.la_valid & LA_CTIME) {
1304                 la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1305                 la->la_valid = LA_CTIME | LA_MTIME;
1306                 rc = mdd_attr_check_set_internal_locked(env, mdd_obj, la,
1307                                                         handle, 0);
1308         }
1309         EXIT;
1310 out_unlock:
1311         mdd_pdo_write_unlock(env, mdd_obj, dlh);
1312 out_trans:
1313         mdd_trans_stop(env, mdd, rc, handle);
1314 out_pending:
1315 #ifdef HAVE_QUOTA_SUPPORT
1316         /* Trigger dqrel for the parent owner.
1317          * If failed, the next call for lquota_chkquota will process it. */
1318         if (quota_opc)
1319                 lquota_adjust(mds_quota_interface_ref, obd, 0, qids, rc,
1320                               quota_opc);
1321 #endif
1322         return rc;
1323 }
1324
1325 /*
1326  * tobj maybe NULL
1327  * has mdd_write_lock on tobj alreay, but not on tgt_pobj yet
1328  */
1329 static int mdd_rt_sanity_check(const struct lu_env *env,
1330                                struct mdd_object *tgt_pobj,
1331                                struct mdd_object *tobj,
1332                                struct md_attr *ma)
1333 {
1334         int rc;
1335         ENTRY;
1336
1337         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
1338                 RETURN(0);
1339
1340         /* XXX: for mdd_rename_tgt, "tobj == NULL" does not mean tobj not
1341          * exist. In fact, tobj must exist, otherwise the call trace will be:
1342          * mdt_reint_rename_tgt -> mdo_name_insert -> ... -> mdd_name_insert.
1343          * When get here, tobj must be NOT NULL, the other case has been
1344          * processed in cmr_rename_tgt before mdd_rename_tgt and enable
1345          * MDS_PERM_BYPASS.
1346          * So check may_delete, but not check nlink of tgt_pobj. */
1347
1348         rc = mdd_may_delete(env, tgt_pobj, tobj, ma, 1, 1);
1349
1350         RETURN(rc);
1351 }
1352
1353 /* Partial rename op on slave MDD */
1354 static int mdd_rename_tgt(const struct lu_env *env,
1355                           struct md_object *pobj, struct md_object *tobj,
1356                           const struct lu_fid *lf, const struct lu_name *lname,
1357                           struct md_attr *ma)
1358 {
1359         const char *name = lname->ln_name;
1360         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
1361         struct mdd_object *mdd_tpobj = md2mdd_obj(pobj);
1362         struct mdd_object *mdd_tobj = md2mdd_obj(tobj);
1363         struct mdd_device *mdd = mdo2mdd(pobj);
1364         struct dynlock_handle *dlh;
1365         struct thandle *handle;
1366 #ifdef HAVE_QUOTA_SUPPORT
1367         struct obd_device *obd = mdd->mdd_obd_dev;
1368         struct obd_export *exp = md_quota(env)->mq_exp;
1369         struct mds_obd *mds = &obd->u.mds;
1370         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1371         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1372         int quota_copc = 0, quota_popc = 0;
1373         int rec_pending[MAXQUOTAS] = { 0, 0 };
1374 #endif
1375         int rc;
1376         ENTRY;
1377
1378         /* XXX: this code won't be used ever:
1379          * DNE uses slightly different approach */
1380         LBUG();
1381
1382 #ifdef HAVE_QUOTA_SUPPORT
1383         if (mds->mds_quota && !tobj) {
1384                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1385
1386                 rc = mdd_la_get(env, mdd_tpobj, la_tmp, BYPASS_CAPA);
1387                 if (!rc) {
1388                         void *data = NULL;
1389                         mdd_data_get(env, mdd_tpobj, &data);
1390                         quota_popc = FSFILT_OP_LINK;
1391                         mdd_quota_wrapper(la_tmp, qpids);
1392                         /* get block quota for target parent */
1393                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1394                                         qpids, rec_pending, 1, NULL,
1395                                         LQUOTA_FLAGS_BLK, data, 1);
1396                 }
1397         }
1398 #endif
1399         handle = mdd_trans_create(env, mdd);
1400         if (IS_ERR(handle))
1401                 GOTO(out_pending, rc = PTR_ERR(handle));
1402
1403         rc = mdd_trans_start(env, mdd, handle);
1404
1405         dlh = mdd_pdo_write_lock(env, mdd_tpobj, name, MOR_TGT_PARENT);
1406         if (dlh == NULL)
1407                 GOTO(out_trans, rc = -ENOMEM);
1408         if (tobj)
1409                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
1410
1411         rc = mdd_rt_sanity_check(env, mdd_tpobj, mdd_tobj, ma);
1412         if (rc)
1413                 GOTO(cleanup, rc);
1414
1415         /*
1416          * If rename_tgt is called then we should just re-insert name with
1417          * correct fid, no need to dec/inc parent nlink if obj is dir.
1418          */
1419         rc = __mdd_index_delete(env, mdd_tpobj, name, 0, handle, BYPASS_CAPA);
1420         if (rc)
1421                 GOTO(cleanup, rc);
1422
1423         rc = __mdd_index_insert_only(env, mdd_tpobj, lf, name, handle,
1424                                      BYPASS_CAPA);
1425         if (rc)
1426                 GOTO(cleanup, rc);
1427
1428         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
1429         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
1430
1431         la->la_valid = LA_CTIME | LA_MTIME;
1432         rc = mdd_attr_check_set_internal_locked(env, mdd_tpobj, la, handle, 0);
1433         if (rc)
1434                 GOTO(cleanup, rc);
1435
1436         /*
1437          * For tobj is remote case cmm layer has processed
1438          * and pass NULL tobj to here. So when tobj is NOT NULL,
1439          * it must be local one.
1440          */
1441         if (tobj && mdd_object_exists(mdd_tobj)) {
1442                 mdo_ref_del(env, mdd_tobj, handle);
1443
1444                 /* Remove dot reference. */
1445                 if (S_ISDIR(ma->ma_attr.la_mode))
1446                         mdo_ref_del(env, mdd_tobj, handle);
1447
1448                 la->la_valid = LA_CTIME;
1449                 rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
1450                 if (rc)
1451                         GOTO(cleanup, rc);
1452
1453                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
1454                 if (rc)
1455                         GOTO(cleanup, rc);
1456
1457 #ifdef HAVE_QUOTA_SUPPORT
1458                 if (mds->mds_quota && ma->ma_valid & MA_INODE &&
1459                     ma->ma_attr.la_nlink == 0 && mdd_tobj->mod_count == 0) {
1460                         quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
1461                         mdd_quota_wrapper(&ma->ma_attr, qcids);
1462                 }
1463 #endif
1464         }
1465         EXIT;
1466 cleanup:
1467         if (tobj)
1468                 mdd_write_unlock(env, mdd_tobj);
1469         mdd_pdo_write_unlock(env, mdd_tpobj, dlh);
1470 out_trans:
1471         if (rc == 0)
1472                 /* Bare EXT record with no RENAME in front of it signifies
1473                    a partial slave op */
1474                 rc = mdd_changelog_ns_store(env, mdd, CL_EXT, 0, mdd_tobj,
1475                                             mdd_tpobj, NULL, lname, handle);
1476
1477         mdd_trans_stop(env, mdd, rc, handle);
1478 out_pending:
1479 #ifdef HAVE_QUOTA_SUPPORT
1480         if (mds->mds_quota) {
1481                 if (quota_popc)
1482                         lquota_pending_commit(mds_quota_interface_ref, obd,
1483                                               qpids, rec_pending, 1);
1484
1485                 if (quota_copc)
1486                         /* Trigger dqrel on the target owner of child.
1487                          * If failed, the next call for lquota_chkquota
1488                          * will process it. */
1489                         lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids,
1490                                       rc, quota_copc);
1491         }
1492 #endif
1493         return rc;
1494 }
1495
1496 /*
1497  * The permission has been checked when obj created, no need check again.
1498  */
1499 static int mdd_cd_sanity_check(const struct lu_env *env,
1500                                struct mdd_object *obj)
1501 {
1502         ENTRY;
1503
1504         /* EEXIST check */
1505         if (!obj || mdd_is_dead_obj(obj))
1506                 RETURN(-ENOENT);
1507
1508         RETURN(0);
1509
1510 }
1511
1512 static int mdd_declare_create_data(const struct lu_env *env,
1513                                    struct mdd_device *mdd,
1514                                    struct mdd_object *obj,
1515                                    int lmm_size,
1516                                    struct thandle *handle)
1517 {
1518         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1519         int            rc;
1520
1521         buf->lb_buf = NULL;
1522         buf->lb_len = lmm_size;
1523         rc = mdo_declare_xattr_set(env, obj, buf, XATTR_NAME_LOV,
1524                                    0, handle);
1525         if (rc)
1526                 return rc;
1527
1528         rc = mdd_declare_lov_objid_update(env, mdd, handle);
1529
1530         return rc;
1531 }
1532
1533 static int mdd_create_data(const struct lu_env *env, struct md_object *pobj,
1534                            struct md_object *cobj, const struct md_op_spec *spec,
1535                            struct md_attr *ma)
1536 {
1537         struct mdd_device *mdd = mdo2mdd(cobj);
1538         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1539         struct mdd_object *son = md2mdd_obj(cobj);
1540         struct lov_mds_md *lmm = NULL;
1541         int                lmm_size = 0;
1542         struct thandle    *handle;
1543         int                rc;
1544         ENTRY;
1545
1546         rc = mdd_cd_sanity_check(env, son);
1547         if (rc)
1548                 RETURN(rc);
1549
1550         if (!md_should_create(spec->sp_cr_flags))
1551                 RETURN(0);
1552         lmm_size = ma->ma_lmm_size;
1553
1554         rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size, spec, ma);
1555         if (rc)
1556                 RETURN(rc);
1557
1558         handle = mdd_trans_create(env, mdd);
1559         if (IS_ERR(handle))
1560                 GOTO(out_free, rc = PTR_ERR(handle));
1561
1562         rc = mdd_declare_create_data(env, mdd, son, lmm_size, handle);
1563         if (rc)
1564                 GOTO(stop, rc);
1565
1566         rc = mdd_trans_start(env, mdd, handle);
1567         if (rc)
1568                 GOTO(stop, rc);
1569
1570         /*
1571          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1572          * Should this be fixed?
1573          */
1574
1575         /* Replay creates has objects already */
1576 #if 0
1577         if (spec->no_create) {
1578                 CDEBUG(D_INFO, "we already have lov ea\n");
1579                 rc = mdd_lov_set_md(env, mdd_pobj, son,
1580                                     (struct lov_mds_md *)spec->u.sp_ea.eadata,
1581                                     spec->u.sp_ea.eadatalen, handle, 0);
1582         } else
1583 #endif
1584                 /* No need mdd_lsm_sanity_check here */
1585                 rc = mdd_lov_set_md(env, mdd_pobj, son, lmm,
1586                                     lmm_size, handle, 0);
1587
1588         if (rc == 0)
1589                rc = mdd_attr_get_internal_locked(env, son, ma);
1590
1591         /* update lov_objid data, must be before transaction stop! */
1592         if (rc == 0)
1593                 mdd_lov_objid_update(mdd, lmm);
1594
1595 stop:
1596         mdd_trans_stop(env, mdd, rc, handle);
1597 out_free:
1598         /* Finish mdd_lov_create() stuff. */
1599         /* if no_create == 0 (not replay), we free lmm allocated by
1600          * mdd_lov_create() */
1601         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
1602         RETURN(rc);
1603 }
1604
1605 /* Get fid from name and parent */
1606 static int
1607 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
1608              const struct lu_name *lname, struct lu_fid* fid, int mask)
1609 {
1610         const char          *name = lname->ln_name;
1611         const struct dt_key *key = (const struct dt_key *)name;
1612         struct mdd_object   *mdd_obj = md2mdd_obj(pobj);
1613         struct mdd_device   *m = mdo2mdd(pobj);
1614         struct dt_object    *dir = mdd_object_child(mdd_obj);
1615         int rc;
1616         ENTRY;
1617
1618         if (unlikely(mdd_is_dead_obj(mdd_obj)))
1619                 RETURN(-ESTALE);
1620
1621         rc = mdd_object_exists(mdd_obj);
1622         if (unlikely(rc == 0))
1623                 RETURN(-ESTALE);
1624         else if (unlikely(rc < 0)) {
1625                 CERROR("Object "DFID" locates on remote server\n",
1626                         PFID(mdo2fid(mdd_obj)));
1627                 RETURN(-EINVAL);
1628         }
1629
1630         /* The common filename length check. */
1631         if (unlikely(lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
1632                 RETURN(-ENAMETOOLONG);
1633
1634         rc = mdd_permission_internal_locked(env, mdd_obj, NULL, mask,
1635                                             MOR_TGT_PARENT);
1636         if (rc)
1637                 RETURN(rc);
1638
1639         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
1640                    dt_try_as_dir(env, dir))) {
1641
1642                 rc = dir->do_index_ops->dio_lookup(env, dir,
1643                                                  (struct dt_rec *)fid, key,
1644                                                  mdd_object_capa(env, mdd_obj));
1645                 if (rc > 0)
1646                         rc = 0;
1647                 else if (rc == 0)
1648                         rc = -ENOENT;
1649         } else
1650                 rc = -ENOTDIR;
1651
1652         RETURN(rc);
1653 }
1654
1655 int mdd_declare_object_initialize(const struct lu_env *env,
1656                                   struct mdd_object *child,
1657                                   struct md_attr *ma,
1658                                   struct thandle *handle)
1659 {
1660         int rc;
1661
1662         rc = mdo_declare_attr_set(env, child, &ma->ma_attr, handle);
1663         if (rc == 0 && S_ISDIR(ma->ma_attr.la_mode)) {
1664                 rc = mdo_declare_index_insert(env, child, mdo2fid(child),
1665                                 dot, handle);
1666                 if (rc == 0)
1667                         rc = mdo_declare_ref_add(env, child, handle);
1668         }
1669         if (rc == 0)
1670                 mdd_declare_links_add(env, child, handle);
1671
1672         return rc;
1673 }
1674
1675 int mdd_object_initialize(const struct lu_env *env, const struct lu_fid *pfid,
1676                           const struct lu_name *lname, struct mdd_object *child,
1677                           struct md_attr *ma, struct thandle *handle,
1678                           const struct md_op_spec *spec)
1679 {
1680         int rc;
1681         ENTRY;
1682
1683         /*
1684          * Update attributes for child.
1685          *
1686          * FIXME:
1687          *  (1) the valid bits should be converted between Lustre and Linux;
1688          *  (2) maybe, the child attributes should be set in OSD when creation.
1689          */
1690
1691         rc = mdd_attr_set_internal(env, child, &ma->ma_attr, handle, 0);
1692         if (rc != 0)
1693                 RETURN(rc);
1694
1695         if (S_ISDIR(ma->ma_attr.la_mode)) {
1696                 /* Add "." and ".." for newly created dir */
1697                 mdo_ref_add(env, child, handle);
1698                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1699                                              dot, handle, BYPASS_CAPA);
1700                 if (rc == 0)
1701                         rc = __mdd_index_insert_only(env, child, pfid,
1702                                                      dotdot, handle,
1703                                                      BYPASS_CAPA);
1704                 if (rc != 0)
1705                         mdo_ref_del(env, child, handle);
1706         }
1707         if (rc == 0)
1708                 mdd_links_add(env, child, pfid, lname, handle, 1);
1709
1710         RETURN(rc);
1711 }
1712
1713 /* has not lock on pobj yet */
1714 static int mdd_create_sanity_check(const struct lu_env *env,
1715                                    struct md_object *pobj,
1716                                    const struct lu_name *lname,
1717                                    struct md_attr *ma,
1718                                    struct md_op_spec *spec)
1719 {
1720         struct mdd_thread_info *info = mdd_env_info(env);
1721         struct lu_attr    *la        = &info->mti_la;
1722         struct lu_fid     *fid       = &info->mti_fid;
1723         struct mdd_object *obj       = md2mdd_obj(pobj);
1724         struct mdd_device *m         = mdo2mdd(pobj);
1725         int lookup                   = spec->sp_cr_lookup;
1726         int rc;
1727         ENTRY;
1728
1729         /* EEXIST check */
1730         if (mdd_is_dead_obj(obj))
1731                 RETURN(-ENOENT);
1732
1733         /*
1734          * In some cases this lookup is not needed - we know before if name
1735          * exists or not because MDT performs lookup for it.
1736          * name length check is done in lookup.
1737          */
1738         if (lookup) {
1739                 /*
1740                  * Check if the name already exist, though it will be checked in
1741                  * _index_insert also, for avoiding rolling back if exists
1742                  * _index_insert.
1743                  */
1744                 rc = __mdd_lookup_locked(env, pobj, lname, fid,
1745                                          MAY_WRITE | MAY_EXEC);
1746                 if (rc != -ENOENT)
1747                         RETURN(rc ? : -EEXIST);
1748         } else {
1749                 /*
1750                  * Check WRITE permission for the parent.
1751                  * EXEC permission have been checked
1752                  * when lookup before create already.
1753                  */
1754                 rc = mdd_permission_internal_locked(env, obj, NULL, MAY_WRITE,
1755                                                     MOR_TGT_PARENT);
1756                 if (rc)
1757                         RETURN(rc);
1758         }
1759
1760         /* sgid check */
1761         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
1762         if (rc != 0)
1763                 RETURN(rc);
1764
1765         if (la->la_mode & S_ISGID) {
1766                 ma->ma_attr.la_gid = la->la_gid;
1767                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1768                         ma->ma_attr.la_mode |= S_ISGID;
1769                         ma->ma_attr.la_valid |= LA_MODE;
1770                 }
1771         }
1772
1773         switch (ma->ma_attr.la_mode & S_IFMT) {
1774         case S_IFLNK: {
1775                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1776
1777                 if (symlen > (1 << m->mdd_dt_conf.ddp_block_shift))
1778                         RETURN(-ENAMETOOLONG);
1779                 else
1780                         RETURN(0);
1781         }
1782         case S_IFDIR:
1783         case S_IFREG:
1784         case S_IFCHR:
1785         case S_IFBLK:
1786         case S_IFIFO:
1787         case S_IFSOCK:
1788                 rc = 0;
1789                 break;
1790         default:
1791                 rc = -EINVAL;
1792                 break;
1793         }
1794         RETURN(rc);
1795 }
1796
1797 static int mdd_declare_create(const struct lu_env *env,
1798                               struct mdd_device *mdd,
1799                               struct mdd_object *p,
1800                               struct mdd_object *c,
1801                               const struct lu_name *name,
1802                               struct md_attr *ma,
1803                               int lmm_size,
1804                               struct thandle *handle,
1805                               const struct md_op_spec *spec)
1806 {
1807         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1808         int            rc = 0;
1809
1810         rc = mdd_declare_object_create_internal(env, p, c, ma, handle, spec);
1811         if (rc)
1812                 GOTO(out, rc);
1813
1814         /* if dir, then can inherit default ACl */
1815         buf->lb_buf = NULL;
1816         buf->lb_len = lmm_size;
1817         if (S_ISDIR(ma->ma_attr.la_mode)) {
1818                 rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_DEFAULT,
1819                                            0, handle);
1820                 if (rc == 0)
1821                         rc = mdo_declare_ref_add(env, p, handle);
1822         }
1823         if (rc)
1824                 GOTO(out, rc);
1825
1826         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_ACCESS,
1827                                    0, handle);
1828         if (rc)
1829                 GOTO(out, rc);
1830
1831         rc = mdd_declare_object_initialize(env, c, ma, handle);
1832         if (rc)
1833                 GOTO(out, rc);
1834
1835         rc = mdo_declare_index_insert(env, p, mdo2fid(c),
1836                                       name->ln_name, handle);
1837         if (rc)
1838                 GOTO(out, rc);
1839
1840         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV,
1841                                    0, handle);
1842         if (rc)
1843                 GOTO(out, rc);
1844
1845         if (S_ISLNK(ma->ma_attr.la_mode)) {
1846                 rc = dt_declare_record_write(env, mdd_object_child(c),
1847                                              strlen(spec->u.sp_symname), 0,
1848                                              handle);
1849                 if (rc)
1850                         GOTO(out, rc);
1851         }
1852         rc = mdo_declare_attr_set(env, p, &ma->ma_attr, handle);
1853         if (rc)
1854                 return rc;
1855
1856         rc = mdd_declare_changelog_store(env, mdd, name, handle);
1857         if (rc)
1858                 return rc;
1859
1860         rc = mdd_declare_lov_objid_update(env, mdd, handle);
1861
1862 out:
1863         return rc;
1864 }
1865
1866
1867 /*
1868  * Create object and insert it into namespace.
1869  */
1870 static int mdd_create(const struct lu_env *env,
1871                       struct md_object *pobj,
1872                       const struct lu_name *lname,
1873                       struct md_object *child,
1874                       struct md_op_spec *spec,
1875                       struct md_attr* ma)
1876 {
1877         struct mdd_thread_info *info = mdd_env_info(env);
1878         struct lu_attr         *la = &info->mti_la_for_fix;
1879         struct md_attr         *ma_acl = &info->mti_ma;
1880         struct mdd_object      *mdd_pobj = md2mdd_obj(pobj);
1881         struct mdd_object      *son = md2mdd_obj(child);
1882         struct mdd_device      *mdd = mdo2mdd(pobj);
1883         struct lu_attr         *attr = &ma->ma_attr;
1884         struct lov_mds_md      *lmm = NULL;
1885         struct thandle         *handle;
1886         struct dynlock_handle  *dlh;
1887         const char             *name = lname->ln_name;
1888         int rc, created = 0, initialized = 0, inserted = 0, lmm_size = 0;
1889         int got_def_acl = 0;
1890 #ifdef HAVE_QUOTA_SUPPORT
1891         struct obd_device *obd = mdd->mdd_obd_dev;
1892         struct obd_export *exp = md_quota(env)->mq_exp;
1893         struct mds_obd *mds = &obd->u.mds;
1894         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1895         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1896         int quota_opc = 0, block_count = 0;
1897         int inode_pending[MAXQUOTAS] = { 0, 0 };
1898         int block_pending[MAXQUOTAS] = { 0, 0 };
1899         int parent_pending[MAXQUOTAS] = { 0, 0 };
1900 #endif
1901         ENTRY;
1902
1903         /*
1904          * Two operations have to be performed:
1905          *
1906          *  - an allocation of a new object (->do_create()), and
1907          *
1908          *  - an insertion into a parent index (->dio_insert()).
1909          *
1910          * Due to locking, operation order is not important, when both are
1911          * successful, *but* error handling cases are quite different:
1912          *
1913          *  - if insertion is done first, and following object creation fails,
1914          *  insertion has to be rolled back, but this operation might fail
1915          *  also leaving us with dangling index entry.
1916          *
1917          *  - if creation is done first, is has to be undone if insertion
1918          *  fails, leaving us with leaked space, which is neither good, nor
1919          *  fatal.
1920          *
1921          * It seems that creation-first is simplest solution, but it is
1922          * sub-optimal in the frequent
1923          *
1924          *         $ mkdir foo
1925          *         $ mkdir foo
1926          *
1927          * case, because second mkdir is bound to create object, only to
1928          * destroy it immediately.
1929          *
1930          * To avoid this follow local file systems that do double lookup:
1931          *
1932          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
1933          *
1934          *     1. create            (mdd_object_create_internal())
1935          *
1936          *     2. insert            (__mdd_index_insert(), lookup again)
1937          */
1938
1939         /* Sanity checks before big job. */
1940         rc = mdd_create_sanity_check(env, pobj, lname, ma, spec);
1941         if (rc)
1942                 RETURN(rc);
1943
1944 #ifdef HAVE_QUOTA_SUPPORT
1945         if (mds->mds_quota) {
1946                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1947
1948                 rc = mdd_la_get(env, mdd_pobj, la_tmp, BYPASS_CAPA);
1949                 if (!rc) {
1950                         int same = 0;
1951
1952                         quota_opc = FSFILT_OP_CREATE;
1953                         mdd_quota_wrapper(&ma->ma_attr, qcids);
1954                         mdd_quota_wrapper(la_tmp, qpids);
1955                         /* get file quota for child */
1956                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1957                                         qcids, inode_pending, 1, NULL, 0, NULL,
1958                                         0);
1959                         switch (ma->ma_attr.la_mode & S_IFMT) {
1960                         case S_IFLNK:
1961                         case S_IFDIR:
1962                                 block_count = 2;
1963                                 break;
1964                         case S_IFREG:
1965                                 block_count = 1;
1966                                 break;
1967                         }
1968                         if (qcids[USRQUOTA] == qpids[USRQUOTA] &&
1969                             qcids[GRPQUOTA] == qpids[GRPQUOTA]) {
1970                                 block_count += 1;
1971                                 same = 1;
1972                         }
1973                         /* get block quota for child and parent */
1974                         if (block_count)
1975                                 lquota_chkquota(mds_quota_interface_ref, obd,
1976                                                 exp, qcids, block_pending,
1977                                                 block_count, NULL,
1978                                                 LQUOTA_FLAGS_BLK, NULL, 0);
1979                         if (!same)
1980                                 lquota_chkquota(mds_quota_interface_ref, obd,
1981                                                 exp, qpids, parent_pending, 1,
1982                                                 NULL, LQUOTA_FLAGS_BLK, NULL,
1983                                                 0);
1984                 }
1985         }
1986 #endif
1987
1988         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_DQACQ_NET))
1989                 GOTO(out_pending, rc = -EINPROGRESS);
1990
1991         /*
1992          * No RPC inside the transaction, so OST objects should be created at
1993          * first.
1994          */
1995         if (S_ISREG(attr->la_mode)) {
1996                 lmm_size = ma->ma_lmm_size;
1997                 rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
1998                                     spec, ma);
1999                 if (rc)
2000                         GOTO(out_pending, rc);
2001         }
2002
2003         if (!S_ISLNK(attr->la_mode)) {
2004                 ma_acl->ma_acl_size = sizeof info->mti_xattr_buf;
2005                 ma_acl->ma_acl = info->mti_xattr_buf;
2006                 ma_acl->ma_need = MA_ACL_DEF;
2007                 ma_acl->ma_valid = 0;
2008
2009                 mdd_read_lock(env, mdd_pobj, MOR_TGT_PARENT);
2010                 rc = mdd_def_acl_get(env, mdd_pobj, ma_acl);
2011                 mdd_read_unlock(env, mdd_pobj);
2012                 if (rc)
2013                         GOTO(out_free, rc);
2014                 else if (ma_acl->ma_valid & MA_ACL_DEF)
2015                         got_def_acl = 1;
2016         }
2017
2018         handle = mdd_trans_create(env, mdd);
2019         if (IS_ERR(handle))
2020                 GOTO(out_free, rc = PTR_ERR(handle));
2021
2022         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, ma,
2023                                 lmm_size, handle, spec);
2024         if (rc)
2025                 GOTO(out_stop, rc);
2026
2027         rc = mdd_trans_start(env, mdd, handle);
2028         if (rc)
2029                 GOTO(out_stop, rc);
2030
2031         dlh = mdd_pdo_write_lock(env, mdd_pobj, name, MOR_TGT_PARENT);
2032         if (dlh == NULL)
2033                 GOTO(out_trans, rc = -ENOMEM);
2034
2035         mdd_write_lock(env, son, MOR_TGT_CHILD);
2036         rc = mdd_object_create_internal(env, mdd_pobj, son, ma, handle, spec);
2037         if (rc) {
2038                 mdd_write_unlock(env, son);
2039                 GOTO(cleanup, rc);
2040         }
2041
2042         created = 1;
2043
2044 #ifdef CONFIG_FS_POSIX_ACL
2045         if (got_def_acl) {
2046                 struct lu_buf *acl_buf = &info->mti_buf;
2047                 acl_buf->lb_buf = ma_acl->ma_acl;
2048                 acl_buf->lb_len = ma_acl->ma_acl_size;
2049
2050                 rc = __mdd_acl_init(env, son, acl_buf, &attr->la_mode, handle);
2051                 if (rc) {
2052                         mdd_write_unlock(env, son);
2053                         GOTO(cleanup, rc);
2054                 } else {
2055                         ma->ma_attr.la_valid |= LA_MODE;
2056                 }
2057         }
2058 #endif
2059
2060         rc = mdd_object_initialize(env, mdo2fid(mdd_pobj), lname,
2061                                    son, ma, handle, spec);
2062         mdd_write_unlock(env, son);
2063         if (rc)
2064                 /*
2065                  * Object has no links, so it will be destroyed when last
2066                  * reference is released. (XXX not now.)
2067                  */
2068                 GOTO(cleanup, rc);
2069
2070         initialized = 1;
2071
2072         rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2073                                 name, S_ISDIR(attr->la_mode), handle,
2074                                 mdd_object_capa(env, mdd_pobj));
2075
2076         if (rc)
2077                 GOTO(cleanup, rc);
2078
2079         inserted = 1;
2080
2081         /* No need mdd_lsm_sanity_check here */
2082         rc = mdd_lov_set_md(env, mdd_pobj, son, lmm, lmm_size, handle, 0);
2083         if (rc) {
2084                 CERROR("error on stripe info copy %d \n", rc);
2085                 GOTO(cleanup, rc);
2086         }
2087         if (lmm && lmm_size > 0) {
2088                 /* Set Lov here, do not get lmm again later */
2089                 if (lmm_size > ma->ma_lmm_size) {
2090                         /* Reply buffer is smaller, need bigger one */
2091                         mdd_max_lmm_buffer(env, lmm_size);
2092                         if (unlikely(info->mti_max_lmm == NULL))
2093                                 GOTO(cleanup, rc = -ENOMEM);
2094                         ma->ma_lmm = info->mti_max_lmm;
2095                         ma->ma_big_lmm_used = 1;
2096                 }
2097                 memcpy(ma->ma_lmm, lmm, lmm_size);
2098                 ma->ma_lmm_size = lmm_size;
2099                 ma->ma_valid |= MA_LOV;
2100         }
2101
2102         if (S_ISLNK(attr->la_mode)) {
2103                 struct md_ucred  *uc = md_ucred(env);
2104                 struct dt_object *dt = mdd_object_child(son);
2105                 const char *target_name = spec->u.sp_symname;
2106                 int sym_len = strlen(target_name);
2107                 const struct lu_buf *buf;
2108                 loff_t pos = 0;
2109
2110                 buf = mdd_buf_get_const(env, target_name, sym_len);
2111                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
2112                                                 mdd_object_capa(env, son),
2113                                                 uc->mu_cap &
2114                                                 CFS_CAP_SYS_RESOURCE_MASK);
2115
2116                 if (rc == sym_len)
2117                         rc = 0;
2118                 else
2119                         GOTO(cleanup, rc = -EFAULT);
2120         }
2121
2122         *la = ma->ma_attr;
2123         la->la_valid = LA_CTIME | LA_MTIME;
2124         rc = mdd_attr_check_set_internal_locked(env, mdd_pobj, la, handle, 0);
2125         if (rc)
2126                 GOTO(cleanup, rc);
2127
2128         /* Return attr back. */
2129         rc = mdd_attr_get_internal_locked(env, son, ma);
2130         EXIT;
2131 cleanup:
2132         if (rc && created) {
2133                 int rc2 = 0;
2134
2135                 if (inserted) {
2136                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2137                                                  S_ISDIR(attr->la_mode),
2138                                                  handle, BYPASS_CAPA);
2139                         if (rc2)
2140                                 CERROR("error can not cleanup destroy %d\n",
2141                                        rc2);
2142                 }
2143
2144                 if (rc2 == 0) {
2145                         mdd_write_lock(env, son, MOR_TGT_CHILD);
2146                         mdo_ref_del(env, son, handle);
2147                         if (initialized && S_ISDIR(attr->la_mode))
2148                                 mdo_ref_del(env, son, handle);
2149                         mdd_write_unlock(env, son);
2150                 }
2151         }
2152
2153         /* update lov_objid data, must be before transaction stop! */
2154         if (rc == 0)
2155                 mdd_lov_objid_update(mdd, lmm);
2156
2157         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
2158 out_trans:
2159         if (rc == 0)
2160                 rc = mdd_changelog_ns_store(env, mdd,
2161                             S_ISDIR(attr->la_mode) ? CL_MKDIR :
2162                             S_ISREG(attr->la_mode) ? CL_CREATE :
2163                             S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2164                             0, son, mdd_pobj, NULL, lname, handle);
2165 out_stop:
2166         mdd_trans_stop(env, mdd, rc, handle);
2167 out_free:
2168         /* finish lov_create stuff, free all temporary data */
2169         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
2170 out_pending:
2171 #ifdef HAVE_QUOTA_SUPPORT
2172         if (quota_opc) {
2173                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2174                                       inode_pending, 0);
2175                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2176                                       block_pending, 1);
2177                 lquota_pending_commit(mds_quota_interface_ref, obd, qpids,
2178                                       parent_pending, 1);
2179                 /* Trigger dqacq on the owner of child and parent. If failed,
2180                  * the next call for lquota_chkquota will process it. */
2181                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2182                               quota_opc);
2183         }
2184 #endif
2185         /* The child object shouldn't be cached anymore */
2186         if (rc)
2187                 cfs_set_bit(LU_OBJECT_HEARD_BANSHEE,
2188                             &child->mo_lu.lo_header->loh_flags);
2189         return rc;
2190 }
2191
2192 /*
2193  * Get locks on parents in proper order
2194  * RETURN: < 0 - error, rename_order if successful
2195  */
2196 enum rename_order {
2197         MDD_RN_SAME,
2198         MDD_RN_SRCTGT,
2199         MDD_RN_TGTSRC
2200 };
2201
2202 static int mdd_rename_order(const struct lu_env *env,
2203                             struct mdd_device *mdd,
2204                             struct mdd_object *src_pobj,
2205                             struct mdd_object *tgt_pobj)
2206 {
2207         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2208         int rc;
2209         ENTRY;
2210
2211         if (src_pobj == tgt_pobj)
2212                 RETURN(MDD_RN_SAME);
2213
2214         /* compared the parent child relationship of src_p&tgt_p */
2215         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2216                 rc = MDD_RN_SRCTGT;
2217         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2218                 rc = MDD_RN_TGTSRC;
2219         } else {
2220                 rc = mdd_is_parent(env, mdd, src_pobj, mdo2fid(tgt_pobj), NULL);
2221                 if (rc == -EREMOTE)
2222                         rc = 0;
2223
2224                 if (rc == 1)
2225                         rc = MDD_RN_TGTSRC;
2226                 else if (rc == 0)
2227                         rc = MDD_RN_SRCTGT;
2228         }
2229
2230         RETURN(rc);
2231 }
2232
2233 /* has not mdd_write{read}_lock on any obj yet. */
2234 static int mdd_rename_sanity_check(const struct lu_env *env,
2235                                    struct mdd_object *src_pobj,
2236                                    struct mdd_object *tgt_pobj,
2237                                    struct mdd_object *sobj,
2238                                    struct mdd_object *tobj,
2239                                    struct md_attr *ma)
2240 {
2241         int rc = 0;
2242         ENTRY;
2243
2244         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
2245                 RETURN(0);
2246
2247         /* XXX: when get here, sobj must NOT be NULL,
2248          * the other case has been processed in cml_rename
2249          * before mdd_rename and enable MDS_PERM_BYPASS. */
2250         LASSERT(sobj);
2251
2252         rc = mdd_may_delete(env, src_pobj, sobj, ma, 1, 0);
2253         if (rc)
2254                 RETURN(rc);
2255
2256         /* XXX: when get here, "tobj == NULL" means tobj must
2257          * NOT exist (neither on remote MDS, such case has been
2258          * processed in cml_rename before mdd_rename and enable
2259          * MDS_PERM_BYPASS).
2260          * So check may_create, but not check may_unlink. */
2261         if (!tobj)
2262                 rc = mdd_may_create(env, tgt_pobj, NULL,
2263                                     (src_pobj != tgt_pobj), 0);
2264         else
2265                 rc = mdd_may_delete(env, tgt_pobj, tobj, ma,
2266                                     (src_pobj != tgt_pobj), 1);
2267
2268         if (!rc && !tobj && (src_pobj != tgt_pobj) &&
2269             S_ISDIR(ma->ma_attr.la_mode))
2270                 rc = __mdd_may_link(env, tgt_pobj);
2271
2272         RETURN(rc);
2273 }
2274
2275 static int mdd_declare_rename(const struct lu_env *env,
2276                               struct mdd_device *mdd,
2277                               struct mdd_object *mdd_spobj,
2278                               struct mdd_object *mdd_tpobj,
2279                               struct mdd_object *mdd_sobj,
2280                               struct mdd_object *mdd_tobj,
2281                               const struct lu_name *sname,
2282                               const struct lu_name *tname,
2283                               struct md_attr *ma,
2284                               struct thandle *handle)
2285 {
2286         int rc;
2287
2288         LASSERT(mdd_spobj);
2289         LASSERT(mdd_tpobj);
2290         LASSERT(mdd_sobj);
2291
2292         /* name from source dir */
2293         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2294         if (rc)
2295                 return rc;
2296
2297         /* .. from source child */
2298         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2299                 /* source child can be directory,
2300                  * counted by source dir's nlink */
2301                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2302                 if (rc)
2303                         return rc;
2304
2305                 rc = mdo_declare_index_delete(env, mdd_sobj, dotdot, handle);
2306                 if (rc)
2307                         return rc;
2308
2309                 rc = mdo_declare_index_insert(env, mdd_sobj, mdo2fid(mdd_tpobj),
2310                                               dotdot, handle);
2311                 if (rc)
2312                         return rc;
2313
2314                 /* new target child can be directory,
2315                  * counted by target dir's nlink */
2316                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2317                 if (rc)
2318                         return rc;
2319
2320         }
2321
2322         rc = mdo_declare_attr_set(env, mdd_spobj, NULL, handle);
2323         if (rc)
2324                 return rc;
2325
2326         rc = mdo_declare_attr_set(env, mdd_sobj, NULL, handle);
2327         if (rc)
2328                 return rc;
2329         mdd_declare_links_add(env, mdd_sobj, handle);
2330         if (rc)
2331                 return rc;
2332
2333         rc = mdo_declare_attr_set(env, mdd_tpobj, NULL, handle);
2334         if (rc)
2335                 return rc;
2336
2337         /* new name */
2338         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2339                         tname->ln_name, handle);
2340         if (rc)
2341                 return rc;
2342
2343         /* name from target dir (old name), we declare it unconditionally
2344          * as mdd_rename() calls delete unconditionally as well. so just
2345          * to balance declarations vs calls to change ... */
2346         rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name, handle);
2347         if (rc)
2348                 return rc;
2349
2350         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2351                 /* delete target child in target parent directory */
2352                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2353                 if (rc)
2354                         return rc;
2355
2356                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2357                         /* target child can be directory,
2358                          * delete "." reference in target child directory */
2359                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2360                         if (rc)
2361                                 return rc;
2362
2363                         /* delete ".." reference in target parent directory */
2364                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2365                         if (rc)
2366                                 return rc;
2367                 }
2368
2369                 rc = mdo_declare_attr_set(env, mdd_tobj, NULL, handle);
2370                 if (rc)
2371                         return rc;
2372
2373                 mdd_declare_links_add(env, mdd_tobj, handle);
2374                 if (rc)
2375                         return rc;
2376
2377                 rc = mdd_declare_finish_unlink(env, mdd_tobj, ma, handle);
2378                 if (rc)
2379                         return rc;
2380         }
2381
2382         rc = mdd_declare_changelog_store(env, mdd, tname, handle);
2383         if (rc)
2384                 return rc;
2385
2386         rc = mdd_declare_changelog_store(env, mdd, sname, handle);
2387         if (rc)
2388                 return rc;
2389
2390         return rc;
2391 }
2392
2393 /* src object can be remote that is why we use only fid and type of object */
2394 static int mdd_rename(const struct lu_env *env,
2395                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2396                       const struct lu_fid *lf, const struct lu_name *lsname,
2397                       struct md_object *tobj, const struct lu_name *ltname,
2398                       struct md_attr *ma)
2399 {
2400         const char *sname = lsname->ln_name;
2401         const char *tname = ltname->ln_name;
2402         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2403         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2404         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2405         struct mdd_device *mdd = mdo2mdd(src_pobj);
2406         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2407         struct mdd_object *mdd_tobj = NULL;
2408         struct dynlock_handle *sdlh, *tdlh;
2409         struct thandle *handle;
2410         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2411         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2412         int is_dir;
2413         int rc, rc2;
2414
2415 #ifdef HAVE_QUOTA_SUPPORT
2416         struct obd_device *obd = mdd->mdd_obd_dev;
2417         struct obd_export *exp = md_quota(env)->mq_exp;
2418         struct mds_obd *mds = &obd->u.mds;
2419         unsigned int qspids[MAXQUOTAS] = { 0, 0 };
2420         unsigned int qtcids[MAXQUOTAS] = { 0, 0 };
2421         unsigned int qtpids[MAXQUOTAS] = { 0, 0 };
2422         int quota_copc = 0, quota_popc = 0;
2423         int rec_pending[MAXQUOTAS] = { 0, 0 };
2424 #endif
2425         ENTRY;
2426
2427         LASSERT(ma->ma_attr.la_mode & S_IFMT);
2428         is_dir = S_ISDIR(ma->ma_attr.la_mode);
2429
2430         if (tobj)
2431                 mdd_tobj = md2mdd_obj(tobj);
2432
2433 #ifdef HAVE_QUOTA_SUPPORT
2434         if (mds->mds_quota) {
2435                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
2436
2437                 rc = mdd_la_get(env, mdd_spobj, la_tmp, BYPASS_CAPA);
2438                 if (!rc) {
2439                         mdd_quota_wrapper(la_tmp, qspids);
2440                         if (!tobj) {
2441                                 rc = mdd_la_get(env, mdd_tpobj, la_tmp,
2442                                                 BYPASS_CAPA);
2443                                 if (!rc) {
2444                                         void *data = NULL;
2445                                         mdd_data_get(env, mdd_tpobj, &data);
2446                                         quota_popc = FSFILT_OP_LINK;
2447                                         mdd_quota_wrapper(la_tmp, qtpids);
2448                                         /* get block quota for target parent */
2449                                         lquota_chkquota(mds_quota_interface_ref,
2450                                                         obd, exp, qtpids,
2451                                                         rec_pending, 1, NULL,
2452                                                         LQUOTA_FLAGS_BLK,
2453                                                         data, 1);
2454                                 }
2455                         }
2456                 }
2457         }
2458 #endif
2459         mdd_sobj = mdd_object_find(env, mdd, lf);
2460
2461         handle = mdd_trans_create(env, mdd);
2462         if (IS_ERR(handle))
2463                 GOTO(out_pending, rc = PTR_ERR(handle));
2464
2465         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
2466                                 mdd_tobj, lsname, ltname, ma, handle);
2467         if (rc)
2468                 GOTO(stop, rc);
2469
2470         rc = mdd_trans_start(env, mdd, handle);
2471         if (rc)
2472                 GOTO(stop, rc);
2473
2474         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2475         rc = mdd_rename_order(env, mdd, mdd_spobj, mdd_tpobj);
2476         if (rc < 0)
2477                 GOTO(cleanup_unlocked, rc);
2478
2479         /* Get locks in determined order */
2480         if (rc == MDD_RN_SAME) {
2481                 sdlh = mdd_pdo_write_lock(env, mdd_spobj,
2482                                           sname, MOR_SRC_PARENT);
2483                 /* check hashes to determine do we need one lock or two */
2484                 if (mdd_name2hash(sname) != mdd_name2hash(tname))
2485                         tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,
2486                                 MOR_TGT_PARENT);
2487                 else
2488                         tdlh = sdlh;
2489         } else if (rc == MDD_RN_SRCTGT) {
2490                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_SRC_PARENT);
2491                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_TGT_PARENT);
2492         } else {
2493                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_SRC_PARENT);
2494                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_TGT_PARENT);
2495         }
2496         if (sdlh == NULL || tdlh == NULL)
2497                 GOTO(cleanup, rc = -ENOMEM);
2498
2499         rc = mdd_rename_sanity_check(env, mdd_spobj, mdd_tpobj,
2500                                      mdd_sobj, mdd_tobj, ma);
2501         if (rc)
2502                 GOTO(cleanup, rc);
2503
2504         /* Remove source name from source directory */
2505         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle,
2506                                 mdd_object_capa(env, mdd_spobj));
2507         if (rc)
2508                 GOTO(cleanup, rc);
2509
2510         /* "mv dir1 dir2" needs "dir1/.." link update */
2511         if (is_dir && mdd_sobj && !lu_fid_eq(spobj_fid, tpobj_fid)) {
2512                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2513                                         mdd_object_capa(env, mdd_sobj));
2514                 if (rc)
2515                         GOTO(fixup_spobj2, rc);
2516
2517                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, dotdot,
2518                                       handle, mdd_object_capa(env, mdd_sobj));
2519                 if (rc)
2520                         GOTO(fixup_spobj, rc);
2521         }
2522
2523         /* Remove target name from target directory
2524          * Here tobj can be remote one, so we do index_delete unconditionally
2525          * and -ENOENT is allowed.
2526          */
2527         rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2528                                 mdd_object_capa(env, mdd_tpobj));
2529         if (rc != 0) {
2530                 if (mdd_tobj) {
2531                         /* tname might been renamed to something else */
2532                         GOTO(fixup_spobj, rc);
2533                 }
2534                 if (rc != -ENOENT)
2535                         GOTO(fixup_spobj, rc);
2536         }
2537
2538         /* Insert new fid with target name into target dir */
2539         rc = __mdd_index_insert(env, mdd_tpobj, lf, tname, is_dir, handle,
2540                                 mdd_object_capa(env, mdd_tpobj));
2541         if (rc)
2542                 GOTO(fixup_tpobj, rc);
2543
2544         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2545         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2546
2547         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
2548         if (mdd_sobj) {
2549                 la->la_valid = LA_CTIME;
2550                 rc = mdd_attr_check_set_internal_locked(env, mdd_sobj, la,
2551                                                         handle, 0);
2552                 if (rc)
2553                         GOTO(fixup_tpobj, rc);
2554         }
2555
2556         /* Remove old target object
2557          * For tobj is remote case cmm layer has processed
2558          * and set tobj to NULL then. So when tobj is NOT NULL,
2559          * it must be local one.
2560          */
2561         if (tobj && mdd_object_exists(mdd_tobj)) {
2562                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
2563                 if (mdd_is_dead_obj(mdd_tobj)) {
2564                         mdd_write_unlock(env, mdd_tobj);
2565                         /* shld not be dead, something is wrong */
2566                         CERROR("tobj is dead, something is wrong\n");
2567                         rc = -EINVAL;
2568                         goto cleanup;
2569                 }
2570                 mdo_ref_del(env, mdd_tobj, handle);
2571
2572                 /* Remove dot reference. */
2573                 if (is_dir)
2574                         mdo_ref_del(env, mdd_tobj, handle);
2575
2576                 la->la_valid = LA_CTIME;
2577                 rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
2578                 if (rc)
2579                         GOTO(fixup_tpobj, rc);
2580
2581                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
2582                 mdd_write_unlock(env, mdd_tobj);
2583                 if (rc)
2584                         GOTO(fixup_tpobj, rc);
2585
2586 #ifdef HAVE_QUOTA_SUPPORT
2587                 if (mds->mds_quota && ma->ma_valid & MA_INODE &&
2588                     ma->ma_attr.la_nlink == 0 && mdd_tobj->mod_count == 0) {
2589                         quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
2590                         mdd_quota_wrapper(&ma->ma_attr, qtcids);
2591                 }
2592 #endif
2593         }
2594
2595         la->la_valid = LA_CTIME | LA_MTIME;
2596         rc = mdd_attr_check_set_internal_locked(env, mdd_spobj, la, handle, 0);
2597         if (rc)
2598                 GOTO(fixup_tpobj, rc);
2599
2600         if (mdd_spobj != mdd_tpobj) {
2601                 la->la_valid = LA_CTIME | LA_MTIME;
2602                 rc = mdd_attr_check_set_internal_locked(env, mdd_tpobj, la,
2603                                                   handle, 0);
2604         }
2605
2606         if (rc == 0 && mdd_sobj) {
2607                 mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
2608                 rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
2609                                       mdo2fid(mdd_tpobj), ltname, handle);
2610                 if (rc == -ENOENT)
2611                         /* Old files might not have EA entry */
2612                         mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
2613                                       lsname, handle, 0);
2614                 mdd_write_unlock(env, mdd_sobj);
2615                 /* We don't fail the transaction if the link ea can't be
2616                    updated -- fid2path will use alternate lookup method. */
2617                 rc = 0;
2618         }
2619
2620         EXIT;
2621
2622 fixup_tpobj:
2623         if (rc) {
2624                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2625                                          BYPASS_CAPA);
2626                 if (rc2)
2627                         CWARN("tp obj fix error %d\n",rc2);
2628
2629                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
2630                     !mdd_is_dead_obj(mdd_tobj)) {
2631                         rc2 = __mdd_index_insert(env, mdd_tpobj,
2632                                          mdo2fid(mdd_tobj), tname,
2633                                          is_dir, handle,
2634                                          BYPASS_CAPA);
2635
2636                         if (rc2)
2637                                 CWARN("tp obj fix error %d\n",rc2);
2638                 }
2639         }
2640
2641 fixup_spobj:
2642         if (rc && is_dir && mdd_sobj) {
2643                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2644                                               BYPASS_CAPA);
2645
2646                 if (rc2)
2647                         CWARN("sp obj dotdot delete error %d\n",rc2);
2648
2649
2650                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid,
2651                                               dotdot, handle, BYPASS_CAPA);
2652                 if (rc2)
2653                         CWARN("sp obj dotdot insert error %d\n",rc2);
2654         }
2655
2656 fixup_spobj2:
2657         if (rc) {
2658                 rc2 = __mdd_index_insert(env, mdd_spobj,
2659                                          lf, sname, is_dir, handle, BYPASS_CAPA);
2660                 if (rc2)
2661                         CWARN("sp obj fix error %d\n",rc2);
2662         }
2663 cleanup:
2664         if (likely(tdlh) && sdlh != tdlh)
2665                 mdd_pdo_write_unlock(env, mdd_tpobj, tdlh);
2666         if (likely(sdlh))
2667                 mdd_pdo_write_unlock(env, mdd_spobj, sdlh);
2668 cleanup_unlocked:
2669         if (rc == 0)
2670                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, 0, mdd_tobj,
2671                                             mdd_spobj, lf, lsname, handle);
2672         if (rc == 0) {
2673                 struct lu_fid zero_fid;
2674                 fid_zero(&zero_fid);
2675                 /* If the rename target exist, The CL_EXT record should save
2676                  * the target fid as tfid, otherwise, use zero fid. LU-543 */
2677                 rc = mdd_changelog_ns_store(env, mdd, CL_EXT, 0, mdd_tobj,
2678                                             mdd_tpobj,
2679                                             mdd_tobj ? NULL : &zero_fid,
2680                                             ltname, handle);
2681         }
2682
2683 stop:
2684         mdd_trans_stop(env, mdd, rc, handle);
2685         if (mdd_sobj)
2686                 mdd_object_put(env, mdd_sobj);
2687 out_pending:
2688 #ifdef HAVE_QUOTA_SUPPORT
2689         if (mds->mds_quota) {
2690                 if (quota_popc)
2691                         lquota_pending_commit(mds_quota_interface_ref, obd,
2692                                               qtpids, rec_pending, 1);
2693
2694                 if (quota_copc) {
2695                         /* Trigger dqrel on the source owner of parent.
2696                          * If failed, the next call for lquota_chkquota will
2697                          * process it. */
2698                         lquota_adjust(mds_quota_interface_ref, obd, 0, qspids, rc,
2699                                       FSFILT_OP_UNLINK_PARTIAL_PARENT);
2700
2701                         /* Trigger dqrel on the target owner of child.
2702                          * If failed, the next call for lquota_chkquota
2703                          * will process it. */
2704                         lquota_adjust(mds_quota_interface_ref, obd, qtcids,
2705                                       qtpids, rc, quota_copc);
2706                 }
2707         }
2708 #endif
2709         return rc;
2710 }
2711
2712 /** enable/disable storing of hardlink info */
2713 int mdd_linkea_enable = 1;
2714 CFS_MODULE_PARM(mdd_linkea_enable, "d", int, 0644,
2715                 "record hardlink info in EAs");
2716
2717 /** Read the link EA into a temp buffer.
2718  * Uses the name_buf since it is generally large.
2719  * \retval IS_ERR err
2720  * \retval ptr to \a lu_buf (always \a mti_big_buf)
2721  */
2722 struct lu_buf *mdd_links_get(const struct lu_env *env,
2723                              struct mdd_object *mdd_obj)
2724 {
2725         struct lu_buf *buf;
2726         struct lustre_capa *capa;
2727         struct link_ea_header *leh;
2728         int rc;
2729
2730         /* First try a small buf */
2731         buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2732         if (buf->lb_buf == NULL)
2733                 return ERR_PTR(-ENOMEM);
2734
2735         capa = mdd_object_capa(env, mdd_obj);
2736         rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2737         if (rc == -ERANGE) {
2738                 /* Buf was too small, figure out what we need. */
2739                 mdd_buf_put(buf);
2740                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2741                 if (rc < 0)
2742                         return ERR_PTR(rc);
2743                 buf = mdd_buf_alloc(env, rc);
2744                 if (buf->lb_buf == NULL)
2745                         return ERR_PTR(-ENOMEM);
2746                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2747         }
2748         if (rc < 0)
2749                 return ERR_PTR(rc);
2750
2751         leh = buf->lb_buf;
2752         if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
2753                 leh->leh_magic = LINK_EA_MAGIC;
2754                 leh->leh_reccount = __swab32(leh->leh_reccount);
2755                 leh->leh_len = __swab64(leh->leh_len);
2756                 /* entries are swabbed by mdd_lee_unpack */
2757         }
2758         if (leh->leh_magic != LINK_EA_MAGIC)
2759                 return ERR_PTR(-EINVAL);
2760         if (leh->leh_reccount == 0)
2761                 return ERR_PTR(-ENODATA);
2762
2763         return buf;
2764 }
2765
2766 /** Pack a link_ea_entry.
2767  * All elements are stored as chars to avoid alignment issues.
2768  * Numbers are always big-endian
2769  * \retval record length
2770  */
2771 static int mdd_lee_pack(struct link_ea_entry *lee, const struct lu_name *lname,
2772                         const struct lu_fid *pfid)
2773 {
2774         struct lu_fid   tmpfid;
2775         int             reclen;
2776
2777         fid_cpu_to_be(&tmpfid, pfid);
2778         memcpy(&lee->lee_parent_fid, &tmpfid, sizeof(tmpfid));
2779         memcpy(lee->lee_name, lname->ln_name, lname->ln_namelen);
2780         reclen = sizeof(struct link_ea_entry) + lname->ln_namelen;
2781
2782         lee->lee_reclen[0] = (reclen >> 8) & 0xff;
2783         lee->lee_reclen[1] = reclen & 0xff;
2784         return reclen;
2785 }
2786
2787 void mdd_lee_unpack(const struct link_ea_entry *lee, int *reclen,
2788                     struct lu_name *lname, struct lu_fid *pfid)
2789 {
2790         *reclen = (lee->lee_reclen[0] << 8) | lee->lee_reclen[1];
2791         memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
2792         fid_be_to_cpu(pfid, pfid);
2793         lname->ln_name = lee->lee_name;
2794         lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
2795 }
2796
2797 /** Add a record to the end of link ea buf */
2798 static int __mdd_links_add(const struct lu_env *env, struct lu_buf *buf,
2799                            const struct lu_fid *pfid,
2800                            const struct lu_name *lname)
2801 {
2802         struct link_ea_header *leh;
2803         struct link_ea_entry *lee;
2804         int reclen;
2805
2806         if (lname == NULL || pfid == NULL)
2807                 return -EINVAL;
2808
2809         /* Make sure our buf is big enough for the new one */
2810         leh = buf->lb_buf;
2811         reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
2812         if (leh->leh_len + reclen > buf->lb_len) {
2813                 if (mdd_buf_grow(env, leh->leh_len + reclen) < 0)
2814                         return -ENOMEM;
2815         }
2816
2817         leh = buf->lb_buf;
2818         lee = buf->lb_buf + leh->leh_len;
2819         reclen = mdd_lee_pack(lee, lname, pfid);
2820         leh->leh_len += reclen;
2821         leh->leh_reccount++;
2822         return 0;
2823 }
2824
2825 static int mdd_declare_links_add(const struct lu_env *env,
2826                                  struct mdd_object *mdd_obj,
2827                                  struct thandle *handle)
2828 {
2829         int rc;
2830
2831         /* XXX: max size? */
2832         rc = mdo_declare_xattr_set(env, mdd_obj,
2833                              mdd_buf_get_const(env, NULL, 4096),
2834                              XATTR_NAME_LINK, 0, handle);
2835
2836         return rc;
2837 }
2838
2839 /* For pathologic linkers, we don't want to spend lots of time scanning the
2840  * link ea.  Limit ourseleves to something reasonable; links not in the EA
2841  * can be looked up via (slower) parent lookup.
2842  */
2843 #define LINKEA_MAX_COUNT 128
2844
2845 static int mdd_links_add(const struct lu_env *env,
2846                          struct mdd_object *mdd_obj,
2847                          const struct lu_fid *pfid,
2848                          const struct lu_name *lname,
2849                          struct thandle *handle, int first)
2850 {
2851         struct lu_buf *buf;
2852         struct link_ea_header *leh;
2853         int rc;
2854         ENTRY;
2855
2856         if (!mdd_linkea_enable)
2857                 RETURN(0);
2858
2859         buf = first ? ERR_PTR(-ENODATA) : mdd_links_get(env, mdd_obj);
2860         if (IS_ERR(buf)) {
2861                 rc = PTR_ERR(buf);
2862                 if (rc != -ENODATA) {
2863                         CERROR("link_ea read failed %d "DFID"\n", rc,
2864                                PFID(mdd_object_fid(mdd_obj)));
2865                         RETURN (rc);
2866                 }
2867                 /* empty EA; start one */
2868                 buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2869                 if (buf->lb_buf == NULL)
2870                         RETURN(-ENOMEM);
2871                 leh = buf->lb_buf;
2872                 leh->leh_magic = LINK_EA_MAGIC;
2873                 leh->leh_len = sizeof(struct link_ea_header);
2874                 leh->leh_reccount = 0;
2875         }
2876
2877         leh = buf->lb_buf;
2878         if (leh->leh_reccount > LINKEA_MAX_COUNT)
2879                 RETURN(-EOVERFLOW);
2880
2881         rc = __mdd_links_add(env, buf, pfid, lname);
2882         if (rc)
2883                 RETURN(rc);
2884
2885         leh = buf->lb_buf;
2886         rc = __mdd_xattr_set(env, mdd_obj,
2887                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2888                              XATTR_NAME_LINK, 0, handle);
2889         if (rc) {
2890                 if (rc == -ENOSPC)
2891                         CDEBUG(D_INODE, "link_ea add failed %d "DFID"\n", rc,
2892                                PFID(mdd_object_fid(mdd_obj)));
2893                 else
2894                         CERROR("link_ea add failed %d "DFID"\n", rc,
2895                                PFID(mdd_object_fid(mdd_obj)));
2896         }
2897
2898         if (buf->lb_len > OBD_ALLOC_BIG)
2899                 /* if we vmalloced a large buffer drop it */
2900                 mdd_buf_put(buf);
2901
2902         RETURN (rc);
2903 }
2904
2905 static int mdd_links_rename(const struct lu_env *env,
2906                             struct mdd_object *mdd_obj,
2907                             const struct lu_fid *oldpfid,
2908                             const struct lu_name *oldlname,
2909                             const struct lu_fid *newpfid,
2910                             const struct lu_name *newlname,
2911                             struct thandle *handle)
2912 {
2913         struct lu_buf  *buf;
2914         struct link_ea_header *leh;
2915         struct link_ea_entry  *lee;
2916         struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
2917         struct lu_fid  *tmpfid = &mdd_env_info(env)->mti_fid;
2918         int reclen = 0;
2919         int count;
2920         int rc, rc2 = 0;
2921         ENTRY;
2922
2923         if (!mdd_linkea_enable)
2924                 RETURN(0);
2925
2926         if (mdd_obj->mod_flags & DEAD_OBJ)
2927                 /* No more links, don't bother */
2928                 RETURN(0);
2929
2930         buf = mdd_links_get(env, mdd_obj);
2931         if (IS_ERR(buf)) {
2932                 rc = PTR_ERR(buf);
2933                 if (rc == -ENODATA)
2934                         CDEBUG(D_INODE, "link_ea read failed %d "DFID"\n",
2935                                rc, PFID(mdd_object_fid(mdd_obj)));
2936                 else
2937                         CERROR("link_ea read failed %d "DFID"\n",
2938                                rc, PFID(mdd_object_fid(mdd_obj)));
2939                 RETURN(rc);
2940         }
2941         leh = buf->lb_buf;
2942         lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
2943
2944         /* Find the old record */
2945         for(count = 0; count < leh->leh_reccount; count++) {
2946                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
2947                 if (tmpname->ln_namelen == oldlname->ln_namelen &&
2948                     lu_fid_eq(tmpfid, oldpfid) &&
2949                     (strncmp(tmpname->ln_name, oldlname->ln_name,
2950                              tmpname->ln_namelen) == 0))
2951                         break;
2952                 lee = (struct link_ea_entry *)((char *)lee + reclen);
2953         }
2954         if ((count + 1) > leh->leh_reccount) {
2955                 CDEBUG(D_INODE, "Old link_ea name '%.*s' not found\n",
2956                        oldlname->ln_namelen, oldlname->ln_name);
2957                 GOTO(out, rc = -ENOENT);
2958         }
2959
2960         /* Remove the old record */
2961         leh->leh_reccount--;
2962         leh->leh_len -= reclen;
2963         memmove(lee, (char *)lee + reclen, (char *)leh + leh->leh_len -
2964                 (char *)lee);
2965
2966         /* If renaming, add the new record */
2967         if (newpfid != NULL) {
2968                 /* if the add fails, we still delete the out-of-date old link */
2969                 rc2 = __mdd_links_add(env, buf, newpfid, newlname);
2970                 leh = buf->lb_buf;
2971         }
2972
2973         rc = __mdd_xattr_set(env, mdd_obj,
2974                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2975                              XATTR_NAME_LINK, 0, handle);
2976
2977 out:
2978         if (rc == 0)
2979                 rc = rc2;
2980         if (rc)
2981                 CDEBUG(D_INODE, "link_ea mv/unlink '%.*s' failed %d "DFID"\n",
2982                        oldlname->ln_namelen, oldlname->ln_name, rc,
2983                        PFID(mdd_object_fid(mdd_obj)));
2984
2985         if (buf->lb_len > OBD_ALLOC_BIG)
2986                 /* if we vmalloced a large buffer drop it */
2987                 mdd_buf_put(buf);
2988
2989         RETURN (rc);
2990 }
2991
2992 const struct md_dir_operations mdd_dir_ops = {
2993         .mdo_is_subdir     = mdd_is_subdir,
2994         .mdo_lookup        = mdd_lookup,
2995         .mdo_create        = mdd_create,
2996         .mdo_rename        = mdd_rename,
2997         .mdo_link          = mdd_link,
2998         .mdo_unlink        = mdd_unlink,
2999         .mdo_lum_lmm_cmp   = mdd_lum_lmm_cmp,
3000         .mdo_name_insert   = mdd_name_insert,
3001         .mdo_name_remove   = mdd_name_remove,
3002         .mdo_rename_tgt    = mdd_rename_tgt,
3003         .mdo_create_data   = mdd_create_data,
3004 };