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