lustre/mdd/mdd_dir.c

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