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