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 lov_mds_md *lmm = NULL;
1565         int                lmm_size = 0;
1566         struct thandle    *handle;
1567         int                rc;
1568         ENTRY;
1569
1570         rc = mdd_cd_sanity_check(env, son);
1571         if (rc)
1572                 RETURN(rc);
1573
1574         if (!md_should_create(spec->sp_cr_flags))
1575                 RETURN(0);
1576         lmm_size = ma->ma_lmm_size;
1577
1578         rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size, spec, ma);
1579         if (rc)
1580                 RETURN(rc);
1581
1582         handle = mdd_trans_create(env, mdd);
1583         if (IS_ERR(handle))
1584                 GOTO(out_free, rc = PTR_ERR(handle));
1585
1586         rc = mdd_declare_create_data(env, mdd, son, lmm_size, handle);
1587         if (rc)
1588                 GOTO(stop, rc);
1589
1590         rc = mdd_trans_start(env, mdd, handle);
1591         if (rc)
1592                 GOTO(stop, rc);
1593
1594         /*
1595          * XXX: Setting the lov ea is not locked but setting the attr is locked?
1596          * Should this be fixed?
1597          */
1598
1599         /* Replay creates has objects already */
1600 #if 0
1601         if (spec->no_create) {
1602                 CDEBUG(D_INFO, "we already have lov ea\n");
1603                 rc = mdd_lov_set_md(env, mdd_pobj, son,
1604                                     (struct lov_mds_md *)spec->u.sp_ea.eadata,
1605                                     spec->u.sp_ea.eadatalen, handle, 0);
1606         } else
1607 #endif
1608                 /* No need mdd_lsm_sanity_check here */
1609                 rc = mdd_lov_set_md(env, mdd_pobj, son, lmm,
1610                                     lmm_size, handle, 0);
1611
1612         if (rc == 0)
1613                rc = mdd_attr_get_internal_locked(env, son, ma);
1614
1615         /* update lov_objid data, must be before transaction stop! */
1616         if (rc == 0)
1617                 mdd_lov_objid_update(mdd, lmm);
1618
1619 stop:
1620         mdd_trans_stop(env, mdd, rc, handle);
1621 out_free:
1622         /* Finish mdd_lov_create() stuff. */
1623         /* if no_create == 0 (not replay), we free lmm allocated by
1624          * mdd_lov_create() */
1625         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
1626         RETURN(rc);
1627 }
1628
1629 /* Get fid from name and parent */
1630 static int
1631 __mdd_lookup(const struct lu_env *env, struct md_object *pobj,
1632              const struct lu_name *lname, struct lu_fid* fid, int mask)
1633 {
1634         const char          *name = lname->ln_name;
1635         const struct dt_key *key = (const struct dt_key *)name;
1636         struct mdd_object   *mdd_obj = md2mdd_obj(pobj);
1637         struct mdd_device   *m = mdo2mdd(pobj);
1638         struct dt_object    *dir = mdd_object_child(mdd_obj);
1639         int rc;
1640         ENTRY;
1641
1642         if (unlikely(mdd_is_dead_obj(mdd_obj)))
1643                 RETURN(-ESTALE);
1644
1645         rc = mdd_object_exists(mdd_obj);
1646         if (unlikely(rc == 0))
1647                 RETURN(-ESTALE);
1648         else if (unlikely(rc < 0)) {
1649                 CERROR("Object "DFID" locates on remote server\n",
1650                         PFID(mdo2fid(mdd_obj)));
1651                 RETURN(-EINVAL);
1652         }
1653
1654         /* The common filename length check. */
1655         if (unlikely(lname->ln_namelen > m->mdd_dt_conf.ddp_max_name_len))
1656                 RETURN(-ENAMETOOLONG);
1657
1658         rc = mdd_permission_internal_locked(env, mdd_obj, NULL, mask,
1659                                             MOR_TGT_PARENT);
1660         if (rc)
1661                 RETURN(rc);
1662
1663         if (likely(S_ISDIR(mdd_object_type(mdd_obj)) &&
1664                    dt_try_as_dir(env, dir))) {
1665
1666                 rc = dir->do_index_ops->dio_lookup(env, dir,
1667                                                  (struct dt_rec *)fid, key,
1668                                                  mdd_object_capa(env, mdd_obj));
1669                 if (rc > 0)
1670                         rc = 0;
1671                 else if (rc == 0)
1672                         rc = -ENOENT;
1673         } else
1674                 rc = -ENOTDIR;
1675
1676         RETURN(rc);
1677 }
1678
1679 int mdd_declare_object_initialize(const struct lu_env *env,
1680                                   struct mdd_object *child,
1681                                   struct md_attr *ma,
1682                                   struct thandle *handle)
1683 {
1684         int rc;
1685
1686         rc = mdo_declare_attr_set(env, child, &ma->ma_attr, handle);
1687         if (rc == 0 && S_ISDIR(ma->ma_attr.la_mode)) {
1688                 rc = mdo_declare_index_insert(env, child, mdo2fid(child),
1689                                 dot, handle);
1690                 if (rc == 0)
1691                         rc = mdo_declare_ref_add(env, child, handle);
1692         }
1693         if (rc == 0)
1694                 mdd_declare_links_add(env, child, handle);
1695
1696         return rc;
1697 }
1698
1699 int mdd_object_initialize(const struct lu_env *env, const struct lu_fid *pfid,
1700                           const struct lu_name *lname, struct mdd_object *child,
1701                           struct md_attr *ma, struct thandle *handle,
1702                           const struct md_op_spec *spec)
1703 {
1704         int rc;
1705         ENTRY;
1706
1707         /*
1708          * Update attributes for child.
1709          *
1710          * FIXME:
1711          *  (1) the valid bits should be converted between Lustre and Linux;
1712          *  (2) maybe, the child attributes should be set in OSD when creation.
1713          */
1714
1715         rc = mdd_attr_set_internal(env, child, &ma->ma_attr, handle, 0);
1716         if (rc != 0)
1717                 RETURN(rc);
1718
1719         if (S_ISDIR(ma->ma_attr.la_mode)) {
1720                 /* Add "." and ".." for newly created dir */
1721                 __mdd_ref_add(env, child, handle);
1722                 rc = __mdd_index_insert_only(env, child, mdo2fid(child),
1723                                              dot, handle, BYPASS_CAPA);
1724                 if (rc == 0)
1725                         rc = __mdd_index_insert_only(env, child, pfid,
1726                                                      dotdot, handle,
1727                                                      BYPASS_CAPA);
1728                 if (rc != 0)
1729                         __mdd_ref_del(env, child, handle, 1);
1730         }
1731         if (rc == 0)
1732                 mdd_links_add(env, child, pfid, lname, handle, 1);
1733
1734         RETURN(rc);
1735 }
1736
1737 /* has not lock on pobj yet */
1738 static int mdd_create_sanity_check(const struct lu_env *env,
1739                                    struct md_object *pobj,
1740                                    const struct lu_name *lname,
1741                                    struct md_attr *ma,
1742                                    struct md_op_spec *spec)
1743 {
1744         struct mdd_thread_info *info = mdd_env_info(env);
1745         struct lu_attr    *la        = &info->mti_la;
1746         struct lu_fid     *fid       = &info->mti_fid;
1747         struct mdd_object *obj       = md2mdd_obj(pobj);
1748         struct mdd_device *m         = mdo2mdd(pobj);
1749         int lookup                   = spec->sp_cr_lookup;
1750         int rc;
1751         ENTRY;
1752
1753         /* EEXIST check */
1754         if (mdd_is_dead_obj(obj))
1755                 RETURN(-ENOENT);
1756
1757         /*
1758          * In some cases this lookup is not needed - we know before if name
1759          * exists or not because MDT performs lookup for it.
1760          * name length check is done in lookup.
1761          */
1762         if (lookup) {
1763                 /*
1764                  * Check if the name already exist, though it will be checked in
1765                  * _index_insert also, for avoiding rolling back if exists
1766                  * _index_insert.
1767                  */
1768                 rc = __mdd_lookup_locked(env, pobj, lname, fid,
1769                                          MAY_WRITE | MAY_EXEC);
1770                 if (rc != -ENOENT)
1771                         RETURN(rc ? : -EEXIST);
1772         } else {
1773                 /*
1774                  * Check WRITE permission for the parent.
1775                  * EXEC permission have been checked
1776                  * when lookup before create already.
1777                  */
1778                 rc = mdd_permission_internal_locked(env, obj, NULL, MAY_WRITE,
1779                                                     MOR_TGT_PARENT);
1780                 if (rc)
1781                         RETURN(rc);
1782         }
1783
1784         /* sgid check */
1785         rc = mdd_la_get(env, obj, la, BYPASS_CAPA);
1786         if (rc != 0)
1787                 RETURN(rc);
1788
1789         if (la->la_mode & S_ISGID) {
1790                 ma->ma_attr.la_gid = la->la_gid;
1791                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1792                         ma->ma_attr.la_mode |= S_ISGID;
1793                         ma->ma_attr.la_valid |= LA_MODE;
1794                 }
1795         }
1796
1797         switch (ma->ma_attr.la_mode & S_IFMT) {
1798         case S_IFLNK: {
1799                 unsigned int symlen = strlen(spec->u.sp_symname) + 1;
1800
1801                 if (symlen > (1 << m->mdd_dt_conf.ddp_block_shift))
1802                         RETURN(-ENAMETOOLONG);
1803                 else
1804                         RETURN(0);
1805         }
1806         case S_IFDIR:
1807         case S_IFREG:
1808         case S_IFCHR:
1809         case S_IFBLK:
1810         case S_IFIFO:
1811         case S_IFSOCK:
1812                 rc = 0;
1813                 break;
1814         default:
1815                 rc = -EINVAL;
1816                 break;
1817         }
1818         RETURN(rc);
1819 }
1820
1821 static int mdd_declare_create(const struct lu_env *env,
1822                               struct mdd_device *mdd,
1823                               struct mdd_object *p,
1824                               struct mdd_object *c,
1825                               const struct lu_name *name,
1826                               struct md_attr *ma,
1827                               int lmm_size,
1828                               struct thandle *handle,
1829                               const struct md_op_spec *spec)
1830 {
1831         struct lu_buf *buf = &mdd_env_info(env)->mti_buf;
1832         int            rc = 0;
1833
1834         rc = mdd_declare_object_create_internal(env, p, c, ma, handle, spec);
1835         if (rc)
1836                 GOTO(out, rc);
1837
1838         /* if dir, then can inherit default ACl */
1839         buf->lb_buf = NULL;
1840         buf->lb_len = lmm_size;
1841         if (S_ISDIR(ma->ma_attr.la_mode)) {
1842                 rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_DEFAULT,
1843                                            0, handle);
1844                 if (rc == 0)
1845                         rc = mdo_declare_ref_add(env, p, handle);
1846         }
1847         if (rc)
1848                 GOTO(out, rc);
1849
1850         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_ACL_ACCESS,
1851                                    0, handle);
1852         if (rc)
1853                 GOTO(out, rc);
1854
1855         rc = mdd_declare_object_initialize(env, c, ma, handle);
1856         if (rc)
1857                 GOTO(out, rc);
1858
1859         rc = mdo_declare_index_insert(env, p, mdo2fid(c),
1860                                       name->ln_name, handle);
1861         if (rc)
1862                 GOTO(out, rc);
1863
1864         rc = mdo_declare_xattr_set(env, c, buf, XATTR_NAME_LOV,
1865                                    0, handle);
1866         if (rc)
1867                 GOTO(out, rc);
1868
1869         if (S_ISLNK(ma->ma_attr.la_mode)) {
1870                 rc = dt_declare_record_write(env, mdd_object_child(c),
1871                                              strlen(spec->u.sp_symname), 0,
1872                                              handle);
1873                 if (rc)
1874                         GOTO(out, rc);
1875         }
1876         rc = mdo_declare_attr_set(env, p, &ma->ma_attr, handle);
1877         if (rc)
1878                 return rc;
1879
1880         rc = mdd_declare_changelog_store(env, mdd, name, handle);
1881         if (rc)
1882                 return rc;
1883
1884         rc = mdd_declare_lov_objid_update(env, mdd, handle);
1885
1886 out:
1887         return rc;
1888 }
1889
1890
1891 /*
1892  * Create object and insert it into namespace.
1893  */
1894 static int mdd_create(const struct lu_env *env,
1895                       struct md_object *pobj,
1896                       const struct lu_name *lname,
1897                       struct md_object *child,
1898                       struct md_op_spec *spec,
1899                       struct md_attr* ma)
1900 {
1901         struct mdd_thread_info *info = mdd_env_info(env);
1902         struct lu_attr         *la = &info->mti_la_for_fix;
1903         struct md_attr         *ma_acl = &info->mti_ma;
1904         struct mdd_object      *mdd_pobj = md2mdd_obj(pobj);
1905         struct mdd_object      *son = md2mdd_obj(child);
1906         struct mdd_device      *mdd = mdo2mdd(pobj);
1907         struct lu_attr         *attr = &ma->ma_attr;
1908         struct lov_mds_md      *lmm = NULL;
1909         struct thandle         *handle;
1910         struct dynlock_handle  *dlh;
1911         const char             *name = lname->ln_name;
1912         int rc, created = 0, initialized = 0, inserted = 0, lmm_size = 0;
1913         int got_def_acl = 0;
1914 #ifdef HAVE_QUOTA_SUPPORT
1915         struct obd_device *obd = mdd->mdd_obd_dev;
1916         struct obd_export *exp = md_quota(env)->mq_exp;
1917         struct mds_obd *mds = &obd->u.mds;
1918         unsigned int qcids[MAXQUOTAS] = { 0, 0 };
1919         unsigned int qpids[MAXQUOTAS] = { 0, 0 };
1920         int quota_opc = 0, block_count = 0;
1921         int inode_pending[MAXQUOTAS] = { 0, 0 };
1922         int block_pending[MAXQUOTAS] = { 0, 0 };
1923         int parent_pending[MAXQUOTAS] = { 0, 0 };
1924 #endif
1925         ENTRY;
1926
1927         /*
1928          * Two operations have to be performed:
1929          *
1930          *  - an allocation of a new object (->do_create()), and
1931          *
1932          *  - an insertion into a parent index (->dio_insert()).
1933          *
1934          * Due to locking, operation order is not important, when both are
1935          * successful, *but* error handling cases are quite different:
1936          *
1937          *  - if insertion is done first, and following object creation fails,
1938          *  insertion has to be rolled back, but this operation might fail
1939          *  also leaving us with dangling index entry.
1940          *
1941          *  - if creation is done first, is has to be undone if insertion
1942          *  fails, leaving us with leaked space, which is neither good, nor
1943          *  fatal.
1944          *
1945          * It seems that creation-first is simplest solution, but it is
1946          * sub-optimal in the frequent
1947          *
1948          *         $ mkdir foo
1949          *         $ mkdir foo
1950          *
1951          * case, because second mkdir is bound to create object, only to
1952          * destroy it immediately.
1953          *
1954          * To avoid this follow local file systems that do double lookup:
1955          *
1956          *     0. lookup -> -EEXIST (mdd_create_sanity_check())
1957          *
1958          *     1. create            (mdd_object_create_internal())
1959          *
1960          *     2. insert            (__mdd_index_insert(), lookup again)
1961          */
1962
1963         /* Sanity checks before big job. */
1964         rc = mdd_create_sanity_check(env, pobj, lname, ma, spec);
1965         if (rc)
1966                 RETURN(rc);
1967
1968 #ifdef HAVE_QUOTA_SUPPORT
1969         if (mds->mds_quota) {
1970                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
1971
1972                 rc = mdd_la_get(env, mdd_pobj, la_tmp, BYPASS_CAPA);
1973                 if (!rc) {
1974                         int same = 0;
1975
1976                         quota_opc = FSFILT_OP_CREATE;
1977                         mdd_quota_wrapper(&ma->ma_attr, qcids);
1978                         mdd_quota_wrapper(la_tmp, qpids);
1979                         /* get file quota for child */
1980                         lquota_chkquota(mds_quota_interface_ref, obd, exp,
1981                                         qcids, inode_pending, 1, NULL, 0, NULL,
1982                                         0);
1983                         switch (ma->ma_attr.la_mode & S_IFMT) {
1984                         case S_IFLNK:
1985                         case S_IFDIR:
1986                                 block_count = 2;
1987                                 break;
1988                         case S_IFREG:
1989                                 block_count = 1;
1990                                 break;
1991                         }
1992                         if (qcids[USRQUOTA] == qpids[USRQUOTA] &&
1993                             qcids[GRPQUOTA] == qpids[GRPQUOTA]) {
1994                                 block_count += 1;
1995                                 same = 1;
1996                         }
1997                         /* get block quota for child and parent */
1998                         if (block_count)
1999                                 lquota_chkquota(mds_quota_interface_ref, obd,
2000                                                 exp, qcids, block_pending,
2001                                                 block_count, NULL,
2002                                                 LQUOTA_FLAGS_BLK, NULL, 0);
2003                         if (!same)
2004                                 lquota_chkquota(mds_quota_interface_ref, obd,
2005                                                 exp, qpids, parent_pending, 1,
2006                                                 NULL, LQUOTA_FLAGS_BLK, NULL,
2007                                                 0);
2008                 }
2009         }
2010 #endif
2011
2012         /*
2013          * No RPC inside the transaction, so OST objects should be created at
2014          * first.
2015          */
2016         if (S_ISREG(attr->la_mode)) {
2017                 lmm_size = ma->ma_lmm_size;
2018                 rc = mdd_lov_create(env, mdd, mdd_pobj, son, &lmm, &lmm_size,
2019                                     spec, ma);
2020                 if (rc)
2021                         GOTO(out_pending, rc);
2022         }
2023
2024         if (!S_ISLNK(attr->la_mode)) {
2025                 ma_acl->ma_acl_size = sizeof info->mti_xattr_buf;
2026                 ma_acl->ma_acl = info->mti_xattr_buf;
2027                 ma_acl->ma_need = MA_ACL_DEF;
2028                 ma_acl->ma_valid = 0;
2029
2030                 mdd_read_lock(env, mdd_pobj, MOR_TGT_PARENT);
2031                 rc = mdd_def_acl_get(env, mdd_pobj, ma_acl);
2032                 mdd_read_unlock(env, mdd_pobj);
2033                 if (rc)
2034                         GOTO(out_free, rc);
2035                 else if (ma_acl->ma_valid & MA_ACL_DEF)
2036                         got_def_acl = 1;
2037         }
2038
2039         handle = mdd_trans_create(env, mdd);
2040         if (IS_ERR(handle))
2041                 GOTO(out_free, rc = PTR_ERR(handle));
2042
2043         rc = mdd_declare_create(env, mdd, mdd_pobj, son, lname, ma,
2044                                 lmm_size, handle, spec);
2045         if (rc)
2046                 GOTO(out_stop, rc);
2047
2048         rc = mdd_trans_start(env, mdd, handle);
2049         if (rc)
2050                 GOTO(out_stop, rc);
2051
2052         dlh = mdd_pdo_write_lock(env, mdd_pobj, name, MOR_TGT_PARENT);
2053         if (dlh == NULL)
2054                 GOTO(out_trans, rc = -ENOMEM);
2055
2056         mdd_write_lock(env, son, MOR_TGT_CHILD);
2057         rc = mdd_object_create_internal(env, mdd_pobj, son, ma, handle, spec);
2058         if (rc) {
2059                 mdd_write_unlock(env, son);
2060                 GOTO(cleanup, rc);
2061         }
2062
2063         created = 1;
2064
2065 #ifdef CONFIG_FS_POSIX_ACL
2066         if (got_def_acl) {
2067                 struct lu_buf *acl_buf = &info->mti_buf;
2068                 acl_buf->lb_buf = ma_acl->ma_acl;
2069                 acl_buf->lb_len = ma_acl->ma_acl_size;
2070
2071                 rc = __mdd_acl_init(env, son, acl_buf, &attr->la_mode, handle);
2072                 if (rc) {
2073                         mdd_write_unlock(env, son);
2074                         GOTO(cleanup, rc);
2075                 } else {
2076                         ma->ma_attr.la_valid |= LA_MODE;
2077                 }
2078         }
2079 #endif
2080
2081         rc = mdd_object_initialize(env, mdo2fid(mdd_pobj), lname,
2082                                    son, ma, handle, spec);
2083         mdd_write_unlock(env, son);
2084         if (rc)
2085                 /*
2086                  * Object has no links, so it will be destroyed when last
2087                  * reference is released. (XXX not now.)
2088                  */
2089                 GOTO(cleanup, rc);
2090
2091         initialized = 1;
2092
2093         rc = __mdd_index_insert(env, mdd_pobj, mdo2fid(son),
2094                                 name, S_ISDIR(attr->la_mode), handle,
2095                                 mdd_object_capa(env, mdd_pobj));
2096
2097         if (rc)
2098                 GOTO(cleanup, rc);
2099
2100         inserted = 1;
2101
2102         /* No need mdd_lsm_sanity_check here */
2103         rc = mdd_lov_set_md(env, mdd_pobj, son, lmm, lmm_size, handle, 0);
2104         if (rc) {
2105                 CERROR("error on stripe info copy %d \n", rc);
2106                 GOTO(cleanup, rc);
2107         }
2108         if (lmm && lmm_size > 0) {
2109                 /* Set Lov here, do not get lmm again later */
2110                 memcpy(ma->ma_lmm, lmm, lmm_size);
2111                 ma->ma_lmm_size = lmm_size;
2112                 ma->ma_valid |= MA_LOV;
2113         }
2114
2115         if (S_ISLNK(attr->la_mode)) {
2116                 struct md_ucred  *uc = md_ucred(env);
2117                 struct dt_object *dt = mdd_object_child(son);
2118                 const char *target_name = spec->u.sp_symname;
2119                 int sym_len = strlen(target_name);
2120                 const struct lu_buf *buf;
2121                 loff_t pos = 0;
2122
2123                 buf = mdd_buf_get_const(env, target_name, sym_len);
2124                 rc = dt->do_body_ops->dbo_write(env, dt, buf, &pos, handle,
2125                                                 mdd_object_capa(env, son),
2126                                                 uc->mu_cap &
2127                                                 CFS_CAP_SYS_RESOURCE_MASK);
2128
2129                 if (rc == sym_len)
2130                         rc = 0;
2131                 else
2132                         GOTO(cleanup, rc = -EFAULT);
2133         }
2134
2135         *la = ma->ma_attr;
2136         la->la_valid = LA_CTIME | LA_MTIME;
2137         rc = mdd_attr_check_set_internal_locked(env, mdd_pobj, la, handle, 0);
2138         if (rc)
2139                 GOTO(cleanup, rc);
2140
2141         /* Return attr back. */
2142         rc = mdd_attr_get_internal_locked(env, son, ma);
2143         EXIT;
2144 cleanup:
2145         if (rc && created) {
2146                 int rc2 = 0;
2147
2148                 if (inserted) {
2149                         rc2 = __mdd_index_delete(env, mdd_pobj, name,
2150                                                  S_ISDIR(attr->la_mode),
2151                                                  handle, BYPASS_CAPA);
2152                         if (rc2)
2153                                 CERROR("error can not cleanup destroy %d\n",
2154                                        rc2);
2155                 }
2156
2157                 if (rc2 == 0) {
2158                         mdd_write_lock(env, son, MOR_TGT_CHILD);
2159                         __mdd_ref_del(env, son, handle, 0);
2160                         if (initialized && S_ISDIR(attr->la_mode))
2161                                 __mdd_ref_del(env, son, handle, 1);
2162                         mdd_write_unlock(env, son);
2163                 }
2164         }
2165
2166         /* update lov_objid data, must be before transaction stop! */
2167         if (rc == 0)
2168                 mdd_lov_objid_update(mdd, lmm);
2169
2170         mdd_pdo_write_unlock(env, mdd_pobj, dlh);
2171 out_trans:
2172         if (rc == 0)
2173                 rc = mdd_changelog_ns_store(env, mdd,
2174                             S_ISDIR(attr->la_mode) ? CL_MKDIR :
2175                             S_ISREG(attr->la_mode) ? CL_CREATE :
2176                             S_ISLNK(attr->la_mode) ? CL_SOFTLINK : CL_MKNOD,
2177                             0, son, mdd_pobj, NULL, lname, handle);
2178 out_stop:
2179         mdd_trans_stop(env, mdd, rc, handle);
2180 out_free:
2181         /* finish lov_create stuff, free all temporary data */
2182         mdd_lov_create_finish(env, mdd, lmm, lmm_size, spec);
2183 out_pending:
2184 #ifdef HAVE_QUOTA_SUPPORT
2185         if (quota_opc) {
2186                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2187                                       inode_pending, 0);
2188                 lquota_pending_commit(mds_quota_interface_ref, obd, qcids,
2189                                       block_pending, 1);
2190                 lquota_pending_commit(mds_quota_interface_ref, obd, qpids,
2191                                       parent_pending, 1);
2192                 /* Trigger dqacq on the owner of child and parent. If failed,
2193                  * the next call for lquota_chkquota will process it. */
2194                 lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
2195                               quota_opc);
2196         }
2197 #endif
2198         return rc;
2199 }
2200
2201 /*
2202  * Get locks on parents in proper order
2203  * RETURN: < 0 - error, rename_order if successful
2204  */
2205 enum rename_order {
2206         MDD_RN_SAME,
2207         MDD_RN_SRCTGT,
2208         MDD_RN_TGTSRC
2209 };
2210
2211 static int mdd_rename_order(const struct lu_env *env,
2212                             struct mdd_device *mdd,
2213                             struct mdd_object *src_pobj,
2214                             struct mdd_object *tgt_pobj)
2215 {
2216         /* order of locking, 1 - tgt-src, 0 - src-tgt*/
2217         int rc;
2218         ENTRY;
2219
2220         if (src_pobj == tgt_pobj)
2221                 RETURN(MDD_RN_SAME);
2222
2223         /* compared the parent child relationship of src_p&tgt_p */
2224         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
2225                 rc = MDD_RN_SRCTGT;
2226         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
2227                 rc = MDD_RN_TGTSRC;
2228         } else {
2229                 rc = mdd_is_parent(env, mdd, src_pobj, mdo2fid(tgt_pobj), NULL);
2230                 if (rc == -EREMOTE)
2231                         rc = 0;
2232
2233                 if (rc == 1)
2234                         rc = MDD_RN_TGTSRC;
2235                 else if (rc == 0)
2236                         rc = MDD_RN_SRCTGT;
2237         }
2238
2239         RETURN(rc);
2240 }
2241
2242 /* has not mdd_write{read}_lock on any obj yet. */
2243 static int mdd_rename_sanity_check(const struct lu_env *env,
2244                                    struct mdd_object *src_pobj,
2245                                    struct mdd_object *tgt_pobj,
2246                                    struct mdd_object *sobj,
2247                                    struct mdd_object *tobj,
2248                                    struct md_attr *ma)
2249 {
2250         int rc = 0;
2251         ENTRY;
2252
2253         if (unlikely(ma->ma_attr_flags & MDS_PERM_BYPASS))
2254                 RETURN(0);
2255
2256         /* XXX: when get here, sobj must NOT be NULL,
2257          * the other case has been processed in cml_rename
2258          * before mdd_rename and enable MDS_PERM_BYPASS. */
2259         LASSERT(sobj);
2260
2261         rc = mdd_may_delete(env, src_pobj, sobj, ma, 1, 0);
2262         if (rc)
2263                 RETURN(rc);
2264
2265         /* XXX: when get here, "tobj == NULL" means tobj must
2266          * NOT exist (neither on remote MDS, such case has been
2267          * processed in cml_rename before mdd_rename and enable
2268          * MDS_PERM_BYPASS).
2269          * So check may_create, but not check may_unlink. */
2270         if (!tobj)
2271                 rc = mdd_may_create(env, tgt_pobj, NULL,
2272                                     (src_pobj != tgt_pobj), 0);
2273         else
2274                 rc = mdd_may_delete(env, tgt_pobj, tobj, ma,
2275                                     (src_pobj != tgt_pobj), 1);
2276
2277         if (!rc && !tobj && (src_pobj != tgt_pobj) &&
2278             S_ISDIR(ma->ma_attr.la_mode))
2279                 rc = __mdd_may_link(env, tgt_pobj);
2280
2281         RETURN(rc);
2282 }
2283
2284 static int mdd_declare_rename(const struct lu_env *env,
2285                               struct mdd_device *mdd,
2286                               struct mdd_object *mdd_spobj,
2287                               struct mdd_object *mdd_tpobj,
2288                               struct mdd_object *mdd_sobj,
2289                               struct mdd_object *mdd_tobj,
2290                               const struct lu_name *sname,
2291                               const struct lu_name *tname,
2292                               struct md_attr *ma,
2293                               struct thandle *handle)
2294 {
2295         int rc;
2296
2297         LASSERT(mdd_spobj);
2298         LASSERT(mdd_tpobj);
2299         LASSERT(mdd_sobj);
2300
2301         /* name from source dir */
2302         rc = mdo_declare_index_delete(env, mdd_spobj, sname->ln_name, handle);
2303         if (rc)
2304                 return rc;
2305
2306         /* .. from source child */
2307         if (S_ISDIR(mdd_object_type(mdd_sobj))) {
2308                 /* source child can be directory,
2309                  * counted by source dir's nlink */
2310                 rc = mdo_declare_ref_del(env, mdd_spobj, handle);
2311                 if (rc)
2312                         return rc;
2313
2314                 rc = mdo_declare_index_delete(env, mdd_sobj, dotdot, handle);
2315                 if (rc)
2316                         return rc;
2317
2318                 rc = mdo_declare_index_insert(env, mdd_sobj, mdo2fid(mdd_tpobj),
2319                                               dotdot, handle);
2320                 if (rc)
2321                         return rc;
2322
2323                 /* new target child can be directory,
2324                  * counted by target dir's nlink */
2325                 rc = mdo_declare_ref_add(env, mdd_tpobj, handle);
2326                 if (rc)
2327                         return rc;
2328
2329         }
2330
2331         rc = mdo_declare_attr_set(env, mdd_spobj, NULL, handle);
2332         if (rc)
2333                 return rc;
2334
2335         rc = mdo_declare_attr_set(env, mdd_sobj, NULL, handle);
2336         if (rc)
2337                 return rc;
2338         mdd_declare_links_add(env, mdd_sobj, handle);
2339         if (rc)
2340                 return rc;
2341
2342         rc = mdo_declare_attr_set(env, mdd_tpobj, NULL, handle);
2343         if (rc)
2344                 return rc;
2345
2346         /* new name */
2347         rc = mdo_declare_index_insert(env, mdd_tpobj, mdo2fid(mdd_sobj),
2348                         tname->ln_name, handle);
2349         if (rc)
2350                 return rc;
2351
2352         /* name from target dir (old name), we declare it unconditionally
2353          * as mdd_rename() calls delete unconditionally as well. so just
2354          * to balance declarations vs calls to change ... */
2355         rc = mdo_declare_index_delete(env, mdd_tpobj, tname->ln_name, handle);
2356         if (rc)
2357                 return rc;
2358
2359         if (mdd_tobj && mdd_object_exists(mdd_tobj)) {
2360                 /* delete target child in target parent directory */
2361                 rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2362                 if (rc)
2363                         return rc;
2364
2365                 if (S_ISDIR(mdd_object_type(mdd_tobj))) {
2366                         /* target child can be directory,
2367                          * delete "." reference in target child directory */
2368                         rc = mdo_declare_ref_del(env, mdd_tobj, handle);
2369                         if (rc)
2370                                 return rc;
2371
2372                         /* delete ".." reference in target parent directory */
2373                         rc = mdo_declare_ref_del(env, mdd_tpobj, handle);
2374                         if (rc)
2375                                 return rc;
2376                 }
2377
2378                 rc = mdo_declare_attr_set(env, mdd_tobj, NULL, handle);
2379                 if (rc)
2380                         return rc;
2381
2382                 mdd_declare_links_add(env, mdd_tobj, handle);
2383                 if (rc)
2384                         return rc;
2385
2386                 rc = mdd_declare_finish_unlink(env, mdd_tobj, ma, handle);
2387                 if (rc)
2388                         return rc;
2389         }
2390
2391         rc = mdd_declare_changelog_store(env, mdd, tname, handle);
2392         if (rc)
2393                 return rc;
2394
2395         rc = mdd_declare_changelog_store(env, mdd, sname, handle);
2396         if (rc)
2397                 return rc;
2398
2399         return rc;
2400 }
2401
2402 /* src object can be remote that is why we use only fid and type of object */
2403 static int mdd_rename(const struct lu_env *env,
2404                       struct md_object *src_pobj, struct md_object *tgt_pobj,
2405                       const struct lu_fid *lf, const struct lu_name *lsname,
2406                       struct md_object *tobj, const struct lu_name *ltname,
2407                       struct md_attr *ma)
2408 {
2409         const char *sname = lsname->ln_name;
2410         const char *tname = ltname->ln_name;
2411         struct lu_attr    *la = &mdd_env_info(env)->mti_la_for_fix;
2412         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj); /* source parent */
2413         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
2414         struct mdd_device *mdd = mdo2mdd(src_pobj);
2415         struct mdd_object *mdd_sobj = NULL;                  /* source object */
2416         struct mdd_object *mdd_tobj = NULL;
2417         struct dynlock_handle *sdlh, *tdlh;
2418         struct thandle *handle;
2419         const struct lu_fid *tpobj_fid = mdo2fid(mdd_tpobj);
2420         const struct lu_fid *spobj_fid = mdo2fid(mdd_spobj);
2421         int is_dir;
2422         int rc, rc2;
2423
2424 #ifdef HAVE_QUOTA_SUPPORT
2425         struct obd_device *obd = mdd->mdd_obd_dev;
2426         struct obd_export *exp = md_quota(env)->mq_exp;
2427         struct mds_obd *mds = &obd->u.mds;
2428         unsigned int qspids[MAXQUOTAS] = { 0, 0 };
2429         unsigned int qtcids[MAXQUOTAS] = { 0, 0 };
2430         unsigned int qtpids[MAXQUOTAS] = { 0, 0 };
2431         int quota_copc = 0, quota_popc = 0;
2432         int rec_pending[MAXQUOTAS] = { 0, 0 };
2433 #endif
2434         ENTRY;
2435
2436         LASSERT(ma->ma_attr.la_mode & S_IFMT);
2437         is_dir = S_ISDIR(ma->ma_attr.la_mode);
2438
2439         if (tobj)
2440                 mdd_tobj = md2mdd_obj(tobj);
2441
2442 #ifdef HAVE_QUOTA_SUPPORT
2443         if (mds->mds_quota) {
2444                 struct lu_attr *la_tmp = &mdd_env_info(env)->mti_la;
2445
2446                 rc = mdd_la_get(env, mdd_spobj, la_tmp, BYPASS_CAPA);
2447                 if (!rc) {
2448                         mdd_quota_wrapper(la_tmp, qspids);
2449                         if (!tobj) {
2450                                 rc = mdd_la_get(env, mdd_tpobj, la_tmp,
2451                                                 BYPASS_CAPA);
2452                                 if (!rc) {
2453                                         void *data = NULL;
2454                                         mdd_data_get(env, mdd_tpobj, &data);
2455                                         quota_popc = FSFILT_OP_LINK;
2456                                         mdd_quota_wrapper(la_tmp, qtpids);
2457                                         /* get block quota for target parent */
2458                                         lquota_chkquota(mds_quota_interface_ref,
2459                                                         obd, exp, qtpids,
2460                                                         rec_pending, 1, NULL,
2461                                                         LQUOTA_FLAGS_BLK,
2462                                                         data, 1);
2463                                 }
2464                         }
2465                 }
2466         }
2467 #endif
2468         mdd_sobj = mdd_object_find(env, mdd, lf);
2469
2470         handle = mdd_trans_create(env, mdd);
2471         if (IS_ERR(handle))
2472                 GOTO(out_pending, rc = PTR_ERR(handle));
2473
2474         rc = mdd_declare_rename(env, mdd, mdd_spobj, mdd_tpobj, mdd_sobj,
2475                                 mdd_tobj, lsname, ltname, ma, handle);
2476         if (rc)
2477                 GOTO(stop, rc);
2478
2479         rc = mdd_trans_start(env, mdd, handle);
2480         if (rc)
2481                 GOTO(stop, rc);
2482
2483         /* FIXME: Should consider tobj and sobj too in rename_lock. */
2484         rc = mdd_rename_order(env, mdd, mdd_spobj, mdd_tpobj);
2485         if (rc < 0)
2486                 GOTO(cleanup_unlocked, rc);
2487
2488         /* Get locks in determined order */
2489         if (rc == MDD_RN_SAME) {
2490                 sdlh = mdd_pdo_write_lock(env, mdd_spobj,
2491                                           sname, MOR_SRC_PARENT);
2492                 /* check hashes to determine do we need one lock or two */
2493                 if (mdd_name2hash(sname) != mdd_name2hash(tname))
2494                         tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,
2495                                 MOR_TGT_PARENT);
2496                 else
2497                         tdlh = sdlh;
2498         } else if (rc == MDD_RN_SRCTGT) {
2499                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_SRC_PARENT);
2500                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_TGT_PARENT);
2501         } else {
2502                 tdlh = mdd_pdo_write_lock(env, mdd_tpobj, tname,MOR_SRC_PARENT);
2503                 sdlh = mdd_pdo_write_lock(env, mdd_spobj, sname,MOR_TGT_PARENT);
2504         }
2505         if (sdlh == NULL || tdlh == NULL)
2506                 GOTO(cleanup, rc = -ENOMEM);
2507
2508         rc = mdd_rename_sanity_check(env, mdd_spobj, mdd_tpobj,
2509                                      mdd_sobj, mdd_tobj, ma);
2510         if (rc)
2511                 GOTO(cleanup, rc);
2512
2513         /* Remove source name from source directory */
2514         rc = __mdd_index_delete(env, mdd_spobj, sname, is_dir, handle,
2515                                 mdd_object_capa(env, mdd_spobj));
2516         if (rc)
2517                 GOTO(cleanup, rc);
2518
2519         /* "mv dir1 dir2" needs "dir1/.." link update */
2520         if (is_dir && mdd_sobj && !lu_fid_eq(spobj_fid, tpobj_fid)) {
2521                 rc = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2522                                         mdd_object_capa(env, mdd_sobj));
2523                 if (rc)
2524                         GOTO(fixup_spobj2, rc);
2525
2526                 rc = __mdd_index_insert_only(env, mdd_sobj, tpobj_fid, dotdot,
2527                                       handle, mdd_object_capa(env, mdd_sobj));
2528                 if (rc)
2529                         GOTO(fixup_spobj, rc);
2530         }
2531
2532         /* Remove target name from target directory
2533          * Here tobj can be remote one, so we do index_delete unconditionally
2534          * and -ENOENT is allowed.
2535          */
2536         rc = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2537                                 mdd_object_capa(env, mdd_tpobj));
2538         if (rc != 0) {
2539                 if (mdd_tobj) {
2540                         /* tname might been renamed to something else */
2541                         GOTO(fixup_spobj, rc);
2542                 }
2543                 if (rc != -ENOENT)
2544                         GOTO(fixup_spobj, rc);
2545         }
2546
2547         /* Insert new fid with target name into target dir */
2548         rc = __mdd_index_insert(env, mdd_tpobj, lf, tname, is_dir, handle,
2549                                 mdd_object_capa(env, mdd_tpobj));
2550         if (rc)
2551                 GOTO(fixup_tpobj, rc);
2552
2553         LASSERT(ma->ma_attr.la_valid & LA_CTIME);
2554         la->la_ctime = la->la_mtime = ma->ma_attr.la_ctime;
2555
2556         /* XXX: mdd_sobj must be local one if it is NOT NULL. */
2557         if (mdd_sobj) {
2558                 la->la_valid = LA_CTIME;
2559                 rc = mdd_attr_check_set_internal_locked(env, mdd_sobj, la,
2560                                                         handle, 0);
2561                 if (rc)
2562                         GOTO(fixup_tpobj, rc);
2563         }
2564
2565         /* Remove old target object
2566          * For tobj is remote case cmm layer has processed
2567          * and set tobj to NULL then. So when tobj is NOT NULL,
2568          * it must be local one.
2569          */
2570         if (tobj && mdd_object_exists(mdd_tobj)) {
2571                 mdd_write_lock(env, mdd_tobj, MOR_TGT_CHILD);
2572                 if (mdd_is_dead_obj(mdd_tobj)) {
2573                         mdd_write_unlock(env, mdd_tobj);
2574                         /* shld not be dead, something is wrong */
2575                         CERROR("tobj is dead, something is wrong\n");
2576                         rc = -EINVAL;
2577                         goto cleanup;
2578                 }
2579                 __mdd_ref_del(env, mdd_tobj, handle, 0);
2580
2581                 /* Remove dot reference. */
2582                 if (is_dir)
2583                         __mdd_ref_del(env, mdd_tobj, handle, 1);
2584
2585                 la->la_valid = LA_CTIME;
2586                 rc = mdd_attr_check_set_internal(env, mdd_tobj, la, handle, 0);
2587                 if (rc)
2588                         GOTO(fixup_tpobj, rc);
2589
2590                 rc = mdd_finish_unlink(env, mdd_tobj, ma, handle);
2591                 mdd_write_unlock(env, mdd_tobj);
2592                 if (rc)
2593                         GOTO(fixup_tpobj, rc);
2594
2595 #ifdef HAVE_QUOTA_SUPPORT
2596                 if (mds->mds_quota && ma->ma_valid & MA_INODE &&
2597                     ma->ma_attr.la_nlink == 0 && mdd_tobj->mod_count == 0) {
2598                         quota_copc = FSFILT_OP_UNLINK_PARTIAL_CHILD;
2599                         mdd_quota_wrapper(&ma->ma_attr, qtcids);
2600                 }
2601 #endif
2602         }
2603
2604         la->la_valid = LA_CTIME | LA_MTIME;
2605         rc = mdd_attr_check_set_internal_locked(env, mdd_spobj, la, handle, 0);
2606         if (rc)
2607                 GOTO(fixup_tpobj, rc);
2608
2609         if (mdd_spobj != mdd_tpobj) {
2610                 la->la_valid = LA_CTIME | LA_MTIME;
2611                 rc = mdd_attr_check_set_internal_locked(env, mdd_tpobj, la,
2612                                                   handle, 0);
2613         }
2614
2615         if (rc == 0 && mdd_sobj) {
2616                 mdd_write_lock(env, mdd_sobj, MOR_SRC_CHILD);
2617                 rc = mdd_links_rename(env, mdd_sobj, mdo2fid(mdd_spobj), lsname,
2618                                       mdo2fid(mdd_tpobj), ltname, handle);
2619                 if (rc == -ENOENT)
2620                         /* Old files might not have EA entry */
2621                         mdd_links_add(env, mdd_sobj, mdo2fid(mdd_spobj),
2622                                       lsname, handle, 0);
2623                 mdd_write_unlock(env, mdd_sobj);
2624                 /* We don't fail the transaction if the link ea can't be
2625                    updated -- fid2path will use alternate lookup method. */
2626                 rc = 0;
2627         }
2628
2629         EXIT;
2630
2631 fixup_tpobj:
2632         if (rc) {
2633                 rc2 = __mdd_index_delete(env, mdd_tpobj, tname, is_dir, handle,
2634                                          BYPASS_CAPA);
2635                 if (rc2)
2636                         CWARN("tp obj fix error %d\n",rc2);
2637
2638                 if (mdd_tobj && mdd_object_exists(mdd_tobj) &&
2639                     !mdd_is_dead_obj(mdd_tobj)) {
2640                         rc2 = __mdd_index_insert(env, mdd_tpobj,
2641                                          mdo2fid(mdd_tobj), tname,
2642                                          is_dir, handle,
2643                                          BYPASS_CAPA);
2644
2645                         if (rc2)
2646                                 CWARN("tp obj fix error %d\n",rc2);
2647                 }
2648         }
2649
2650 fixup_spobj:
2651         if (rc && is_dir && mdd_sobj) {
2652                 rc2 = __mdd_index_delete_only(env, mdd_sobj, dotdot, handle,
2653                                               BYPASS_CAPA);
2654
2655                 if (rc2)
2656                         CWARN("sp obj dotdot delete error %d\n",rc2);
2657
2658
2659                 rc2 = __mdd_index_insert_only(env, mdd_sobj, spobj_fid,
2660                                               dotdot, handle, BYPASS_CAPA);
2661                 if (rc2)
2662                         CWARN("sp obj dotdot insert error %d\n",rc2);
2663         }
2664
2665 fixup_spobj2:
2666         if (rc) {
2667                 rc2 = __mdd_index_insert(env, mdd_spobj,
2668                                          lf, sname, is_dir, handle, BYPASS_CAPA);
2669                 if (rc2)
2670                         CWARN("sp obj fix error %d\n",rc2);
2671         }
2672 cleanup:
2673         if (likely(tdlh) && sdlh != tdlh)
2674                 mdd_pdo_write_unlock(env, mdd_tpobj, tdlh);
2675         if (likely(sdlh))
2676                 mdd_pdo_write_unlock(env, mdd_spobj, sdlh);
2677 cleanup_unlocked:
2678         if (rc == 0)
2679                 rc = mdd_changelog_ns_store(env, mdd, CL_RENAME, 0, mdd_tobj,
2680                                             mdd_spobj, lf, lsname, handle);
2681         if (rc == 0) {
2682                 struct lu_fid zero_fid;
2683                 fid_zero(&zero_fid);
2684                 /* If the rename target exist, The CL_EXT record should save
2685                  * the target fid as tfid, otherwise, use zero fid. LU-543 */
2686                 rc = mdd_changelog_ns_store(env, mdd, CL_EXT, 0, mdd_tobj,
2687                                             mdd_tpobj,
2688                                             mdd_tobj ? NULL : &zero_fid,
2689                                             ltname, handle);
2690         }
2691
2692 stop:
2693         mdd_trans_stop(env, mdd, rc, handle);
2694         if (mdd_sobj)
2695                 mdd_object_put(env, mdd_sobj);
2696 out_pending:
2697 #ifdef HAVE_QUOTA_SUPPORT
2698         if (mds->mds_quota) {
2699                 if (quota_popc)
2700                         lquota_pending_commit(mds_quota_interface_ref, obd,
2701                                               qtpids, rec_pending, 1);
2702
2703                 if (quota_copc) {
2704                         /* Trigger dqrel on the source owner of parent.
2705                          * If failed, the next call for lquota_chkquota will
2706                          * process it. */
2707                         lquota_adjust(mds_quota_interface_ref, obd, 0, qspids, rc,
2708                                       FSFILT_OP_UNLINK_PARTIAL_PARENT);
2709
2710                         /* Trigger dqrel on the target owner of child.
2711                          * If failed, the next call for lquota_chkquota
2712                          * will process it. */
2713                         lquota_adjust(mds_quota_interface_ref, obd, qtcids,
2714                                       qtpids, rc, quota_copc);
2715                 }
2716         }
2717 #endif
2718         return rc;
2719 }
2720
2721 /** enable/disable storing of hardlink info */
2722 int mdd_linkea_enable = 1;
2723 CFS_MODULE_PARM(mdd_linkea_enable, "d", int, 0644,
2724                 "record hardlink info in EAs");
2725
2726 /** Read the link EA into a temp buffer.
2727  * Uses the name_buf since it is generally large.
2728  * \retval IS_ERR err
2729  * \retval ptr to \a lu_buf (always \a mti_big_buf)
2730  */
2731 struct lu_buf *mdd_links_get(const struct lu_env *env,
2732                              struct mdd_object *mdd_obj)
2733 {
2734         struct lu_buf *buf;
2735         struct lustre_capa *capa;
2736         struct link_ea_header *leh;
2737         int rc;
2738
2739         /* First try a small buf */
2740         buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2741         if (buf->lb_buf == NULL)
2742                 return ERR_PTR(-ENOMEM);
2743
2744         capa = mdd_object_capa(env, mdd_obj);
2745         rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2746         if (rc == -ERANGE) {
2747                 /* Buf was too small, figure out what we need. */
2748                 mdd_buf_put(buf);
2749                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2750                 if (rc < 0)
2751                         return ERR_PTR(rc);
2752                 buf = mdd_buf_alloc(env, rc);
2753                 if (buf->lb_buf == NULL)
2754                         return ERR_PTR(-ENOMEM);
2755                 rc = mdo_xattr_get(env, mdd_obj, buf, XATTR_NAME_LINK, capa);
2756         }
2757         if (rc < 0)
2758                 return ERR_PTR(rc);
2759
2760         leh = buf->lb_buf;
2761         if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
2762                 leh->leh_magic = LINK_EA_MAGIC;
2763                 leh->leh_reccount = __swab32(leh->leh_reccount);
2764                 leh->leh_len = __swab64(leh->leh_len);
2765                 /* entries are swabbed by mdd_lee_unpack */
2766         }
2767         if (leh->leh_magic != LINK_EA_MAGIC)
2768                 return ERR_PTR(-EINVAL);
2769         if (leh->leh_reccount == 0)
2770                 return ERR_PTR(-ENODATA);
2771
2772         return buf;
2773 }
2774
2775 /** Pack a link_ea_entry.
2776  * All elements are stored as chars to avoid alignment issues.
2777  * Numbers are always big-endian
2778  * \retval record length
2779  */
2780 static int mdd_lee_pack(struct link_ea_entry *lee, const struct lu_name *lname,
2781                         const struct lu_fid *pfid)
2782 {
2783         struct lu_fid   tmpfid;
2784         int             reclen;
2785
2786         fid_cpu_to_be(&tmpfid, pfid);
2787         memcpy(&lee->lee_parent_fid, &tmpfid, sizeof(tmpfid));
2788         memcpy(lee->lee_name, lname->ln_name, lname->ln_namelen);
2789         reclen = sizeof(struct link_ea_entry) + lname->ln_namelen;
2790
2791         lee->lee_reclen[0] = (reclen >> 8) & 0xff;
2792         lee->lee_reclen[1] = reclen & 0xff;
2793         return reclen;
2794 }
2795
2796 void mdd_lee_unpack(const struct link_ea_entry *lee, int *reclen,
2797                     struct lu_name *lname, struct lu_fid *pfid)
2798 {
2799         *reclen = (lee->lee_reclen[0] << 8) | lee->lee_reclen[1];
2800         memcpy(pfid, &lee->lee_parent_fid, sizeof(*pfid));
2801         fid_be_to_cpu(pfid, pfid);
2802         lname->ln_name = lee->lee_name;
2803         lname->ln_namelen = *reclen - sizeof(struct link_ea_entry);
2804 }
2805
2806 /** Add a record to the end of link ea buf */
2807 static int __mdd_links_add(const struct lu_env *env, struct lu_buf *buf,
2808                            const struct lu_fid *pfid,
2809                            const struct lu_name *lname)
2810 {
2811         struct link_ea_header *leh;
2812         struct link_ea_entry *lee;
2813         int reclen;
2814
2815         if (lname == NULL || pfid == NULL)
2816                 return -EINVAL;
2817
2818         /* Make sure our buf is big enough for the new one */
2819         leh = buf->lb_buf;
2820         reclen = lname->ln_namelen + sizeof(struct link_ea_entry);
2821         if (leh->leh_len + reclen > buf->lb_len) {
2822                 if (mdd_buf_grow(env, leh->leh_len + reclen) < 0)
2823                         return -ENOMEM;
2824         }
2825
2826         leh = buf->lb_buf;
2827         lee = buf->lb_buf + leh->leh_len;
2828         reclen = mdd_lee_pack(lee, lname, pfid);
2829         leh->leh_len += reclen;
2830         leh->leh_reccount++;
2831         return 0;
2832 }
2833
2834 static int mdd_declare_links_add(const struct lu_env *env,
2835                                  struct mdd_object *mdd_obj,
2836                                  struct thandle *handle)
2837 {
2838         int rc;
2839
2840         /* XXX: max size? */
2841         rc = mdo_declare_xattr_set(env, mdd_obj,
2842                              mdd_buf_get_const(env, NULL, 4096),
2843                              XATTR_NAME_LINK, 0, handle);
2844
2845         return rc;
2846 }
2847
2848 /* For pathologic linkers, we don't want to spend lots of time scanning the
2849  * link ea.  Limit ourseleves to something reasonable; links not in the EA
2850  * can be looked up via (slower) parent lookup.
2851  */
2852 #define LINKEA_MAX_COUNT 128
2853
2854 static int mdd_links_add(const struct lu_env *env,
2855                          struct mdd_object *mdd_obj,
2856                          const struct lu_fid *pfid,
2857                          const struct lu_name *lname,
2858                          struct thandle *handle, int first)
2859 {
2860         struct lu_buf *buf;
2861         struct link_ea_header *leh;
2862         int rc;
2863         ENTRY;
2864
2865         if (!mdd_linkea_enable)
2866                 RETURN(0);
2867
2868         buf = first ? ERR_PTR(-ENODATA) : mdd_links_get(env, mdd_obj);
2869         if (IS_ERR(buf)) {
2870                 rc = PTR_ERR(buf);
2871                 if (rc != -ENODATA) {
2872                         CERROR("link_ea read failed %d "DFID"\n", rc,
2873                                PFID(mdd_object_fid(mdd_obj)));
2874                         RETURN (rc);
2875                 }
2876                 /* empty EA; start one */
2877                 buf = mdd_buf_alloc(env, CFS_PAGE_SIZE);
2878                 if (buf->lb_buf == NULL)
2879                         RETURN(-ENOMEM);
2880                 leh = buf->lb_buf;
2881                 leh->leh_magic = LINK_EA_MAGIC;
2882                 leh->leh_len = sizeof(struct link_ea_header);
2883                 leh->leh_reccount = 0;
2884         }
2885
2886         leh = buf->lb_buf;
2887         if (leh->leh_reccount > LINKEA_MAX_COUNT)
2888                 RETURN(-EOVERFLOW);
2889
2890         rc = __mdd_links_add(env, buf, pfid, lname);
2891         if (rc)
2892                 RETURN(rc);
2893
2894         leh = buf->lb_buf;
2895         rc = __mdd_xattr_set(env, mdd_obj,
2896                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2897                              XATTR_NAME_LINK, 0, handle);
2898         if (rc) {
2899                 if (rc == -ENOSPC)
2900                         CDEBUG(D_INODE, "link_ea add failed %d "DFID"\n", rc,
2901                                PFID(mdd_object_fid(mdd_obj)));
2902                 else
2903                         CERROR("link_ea add failed %d "DFID"\n", rc,
2904                                PFID(mdd_object_fid(mdd_obj)));
2905         }
2906
2907         if (buf->lb_len > OBD_ALLOC_BIG)
2908                 /* if we vmalloced a large buffer drop it */
2909                 mdd_buf_put(buf);
2910
2911         RETURN (rc);
2912 }
2913
2914 static int mdd_links_rename(const struct lu_env *env,
2915                             struct mdd_object *mdd_obj,
2916                             const struct lu_fid *oldpfid,
2917                             const struct lu_name *oldlname,
2918                             const struct lu_fid *newpfid,
2919                             const struct lu_name *newlname,
2920                             struct thandle *handle)
2921 {
2922         struct lu_buf  *buf;
2923         struct link_ea_header *leh;
2924         struct link_ea_entry  *lee;
2925         struct lu_name *tmpname = &mdd_env_info(env)->mti_name;
2926         struct lu_fid  *tmpfid = &mdd_env_info(env)->mti_fid;
2927         int reclen = 0;
2928         int count;
2929         int rc, rc2 = 0;
2930         ENTRY;
2931
2932         if (!mdd_linkea_enable)
2933                 RETURN(0);
2934
2935         if (mdd_obj->mod_flags & DEAD_OBJ)
2936                 /* No more links, don't bother */
2937                 RETURN(0);
2938
2939         buf = mdd_links_get(env, mdd_obj);
2940         if (IS_ERR(buf)) {
2941                 rc = PTR_ERR(buf);
2942                 if (rc == -ENODATA)
2943                         CDEBUG(D_INODE, "link_ea read failed %d "DFID"\n",
2944                                rc, PFID(mdd_object_fid(mdd_obj)));
2945                 else
2946                         CERROR("link_ea read failed %d "DFID"\n",
2947                                rc, PFID(mdd_object_fid(mdd_obj)));
2948                 RETURN(rc);
2949         }
2950         leh = buf->lb_buf;
2951         lee = (struct link_ea_entry *)(leh + 1); /* link #0 */
2952
2953         /* Find the old record */
2954         for(count = 0; count < leh->leh_reccount; count++) {
2955                 mdd_lee_unpack(lee, &reclen, tmpname, tmpfid);
2956                 if (tmpname->ln_namelen == oldlname->ln_namelen &&
2957                     lu_fid_eq(tmpfid, oldpfid) &&
2958                     (strncmp(tmpname->ln_name, oldlname->ln_name,
2959                              tmpname->ln_namelen) == 0))
2960                         break;
2961                 lee = (struct link_ea_entry *)((char *)lee + reclen);
2962         }
2963         if ((count + 1) > leh->leh_reccount) {
2964                 CDEBUG(D_INODE, "Old link_ea name '%.*s' not found\n",
2965                        oldlname->ln_namelen, oldlname->ln_name);
2966                 GOTO(out, rc = -ENOENT);
2967         }
2968
2969         /* Remove the old record */
2970         leh->leh_reccount--;
2971         leh->leh_len -= reclen;
2972         memmove(lee, (char *)lee + reclen, (char *)leh + leh->leh_len -
2973                 (char *)lee);
2974
2975         /* If renaming, add the new record */
2976         if (newpfid != NULL) {
2977                 /* if the add fails, we still delete the out-of-date old link */
2978                 rc2 = __mdd_links_add(env, buf, newpfid, newlname);
2979                 leh = buf->lb_buf;
2980         }
2981
2982         rc = __mdd_xattr_set(env, mdd_obj,
2983                              mdd_buf_get_const(env, buf->lb_buf, leh->leh_len),
2984                              XATTR_NAME_LINK, 0, handle);
2985
2986 out:
2987         if (rc == 0)
2988                 rc = rc2;
2989         if (rc)
2990                 CDEBUG(D_INODE, "link_ea mv/unlink '%.*s' failed %d "DFID"\n",
2991                        oldlname->ln_namelen, oldlname->ln_name, rc,
2992                        PFID(mdd_object_fid(mdd_obj)));
2993
2994         if (buf->lb_len > OBD_ALLOC_BIG)
2995                 /* if we vmalloced a large buffer drop it */
2996                 mdd_buf_put(buf);
2997
2998         RETURN (rc);
2999 }
3000
3001 const struct md_dir_operations mdd_dir_ops = {
3002         .mdo_is_subdir     = mdd_is_subdir,
3003         .mdo_lookup        = mdd_lookup,
3004         .mdo_create        = mdd_create,
3005         .mdo_rename        = mdd_rename,
3006         .mdo_link          = mdd_link,
3007         .mdo_unlink        = mdd_unlink,
3008         .mdo_lum_lmm_cmp   = mdd_lum_lmm_cmp,
3009         .mdo_name_insert   = mdd_name_insert,
3010         .mdo_name_remove   = mdd_name_remove,
3011         .mdo_rename_tgt    = mdd_rename_tgt,
3012         .mdo_create_data   = mdd_create_data,
3013 };