lustre/mdd/mdd_dir.c

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