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