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