lustre/mdd/mdd_dir.c

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