lustre/mdd/mdd_dir.c

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