lustre/mdd/mdd_handler.c

   1 /* -*- MODE: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  mdd/mdd_handler.c
   5  *  Lustre Metadata Server (mdd) routines
   6  *
   7  *  Copyright (C) 2006 Cluster File Systems, Inc.
   8  *   Author: Wang Di <wangdi@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28 #ifndef EXPORT_SYMTAB
  29 # define EXPORT_SYMTAB
  30 #endif
  31 #define DEBUG_SUBSYSTEM S_MDS
  32
  33 #include <linux/module.h>
  34 #include <linux/jbd.h>
  35
  36 #include <obd.h>
  37 #include <obd_class.h>
  38 #include <lustre_ver.h>
  39 #include <obd_support.h>
  40 #include <lprocfs_status.h>
  41
  42 #include <linux/ldiskfs_fs.h>
  43 #include <lustre_mds.h>
  44 #include <lu_object.h>
  45 #include <md_object.h>
  46 #include <dt_object.h>
  47
  48 #include "mdd_internal.h"
  49
  50
  51 static struct thandle* mdd_trans_start(const struct lu_context *ctxt,
  52                                        struct mdd_device *);
  53 static void mdd_trans_stop(const struct lu_context *ctxt,
  54                            struct mdd_device *mdd, int rc,
  55                            struct thandle *handle);
  56 static struct dt_object* mdd_object_child(struct mdd_object *o);
  57 static void __mdd_ref_add(const struct lu_context *ctxt, struct mdd_object *obj,
  58                           struct thandle *handle);
  59 static void __mdd_ref_del(const struct lu_context *ctxt, struct mdd_object *obj,
  60                           struct thandle *handle);
  61 static int mdd_lookup(const struct lu_context *ctxt, struct md_object *pobj,
  62                       const char *name, struct lu_fid* fid);
  63
  64 static struct md_object_operations mdd_obj_ops;
  65 static struct md_dir_operations    mdd_dir_ops;
  66 static struct lu_object_operations mdd_lu_obj_ops;
  67
  68 static struct lu_context_key       mdd_thread_key;
  69
  70 static const char *mdd_root_dir_name = "root";
  71 static const char dot[] = ".";
  72 static const char dotdot[] = "..";
  73
  74
  75 struct mdd_thread_info *mdd_ctx_info(const struct lu_context *ctx)
  76 {
  77         struct mdd_thread_info *info;
  78
  79         info = lu_context_key_get(ctx, &mdd_thread_key);
  80         LASSERT(info != NULL);
  81         return info;
  82 }
  83
  84 static struct lu_object *mdd_object_alloc(const struct lu_context *ctxt,
  85                                           const struct lu_object_header *hdr,
  86                                           struct lu_device *d)
  87 {
  88         struct mdd_object *mdd_obj;
  89
  90         OBD_ALLOC_PTR(mdd_obj);
  91         if (mdd_obj != NULL) {
  92                 struct lu_object *o;
  93
  94                 o = mdd2lu_obj(mdd_obj);
  95                 lu_object_init(o, NULL, d);
  96                 mdd_obj->mod_obj.mo_ops = &mdd_obj_ops;
  97                 mdd_obj->mod_obj.mo_dir_ops = &mdd_dir_ops;
  98                 atomic_set(&mdd_obj->mod_count, 0);
  99                 o->lo_ops = &mdd_lu_obj_ops;
 100                 return o;
 101         } else {
 102                 return NULL;
 103         }
 104 }
 105
 106 static int mdd_object_init(const struct lu_context *ctxt, struct lu_object *o)
 107 {
 108         struct mdd_device *d = lu2mdd_dev(o->lo_dev);
 109         struct lu_object  *below;
 110         struct lu_device  *under;
 111         ENTRY;
 112
 113         under = &d->mdd_child->dd_lu_dev;
 114         below = under->ld_ops->ldo_object_alloc(ctxt, o->lo_header, under);
 115
 116         if (below == NULL)
 117                 RETURN(-ENOMEM);
 118
 119         lu_object_add(o, below);
 120         RETURN(0);
 121 }
 122
 123 static int mdd_get_flags(const struct lu_context *ctxt, struct mdd_object *obj);
 124
 125 static int mdd_object_start(const struct lu_context *ctxt, struct lu_object *o)
 126 {
 127         if (lu_object_exists(o))
 128                 return mdd_get_flags(ctxt, lu2mdd_obj(o));
 129         else
 130                 return 0;
 131 }
 132
 133 static void mdd_object_free(const struct lu_context *ctxt, struct lu_object *o)
 134 {
 135         struct mdd_object *mdd = lu2mdd_obj(o);
 136
 137         lu_object_fini(o);
 138         OBD_FREE_PTR(mdd);
 139 }
 140
 141 struct mdd_object *mdd_object_find(const struct lu_context *ctxt,
 142                                    struct mdd_device *d,
 143                                    const struct lu_fid *f)
 144 {
 145         struct lu_object *o;
 146         struct mdd_object *m;
 147         ENTRY;
 148
 149         o = lu_object_find(ctxt, d->mdd_md_dev.md_lu_dev.ld_site, f);
 150         if (IS_ERR(o))
 151                 m = (struct mdd_object *)o;
 152         else
 153                 m = lu2mdd_obj(lu_object_locate(o->lo_header,
 154                                  d->mdd_md_dev.md_lu_dev.ld_type));
 155         RETURN(m);
 156 }
 157
 158 static inline void mdd_object_put(const struct lu_context *ctxt,
 159                                   struct mdd_object *o)
 160 {
 161         lu_object_put(ctxt, &o->mod_obj.mo_lu);
 162 }
 163
 164 static inline int mdd_is_immutable(struct mdd_object *obj)
 165 {
 166         return obj->mod_flags & IMMUTE_OBJ;
 167 }
 168
 169 static inline int mdd_is_append(struct mdd_object *obj)
 170 {
 171         return obj->mod_flags & APPEND_OBJ;
 172 }
 173
 174 static void mdd_set_dead_obj(struct mdd_object *obj)
 175 {
 176         if (obj)
 177                 obj->mod_flags |= DEAD_OBJ;
 178 }
 179
 180 static int mdd_is_dead_obj(struct mdd_object *obj)
 181 {
 182         return obj && obj->mod_flags & DEAD_OBJ;
 183 }
 184
 185 /*Check whether it may create the cobj under the pobj*/
 186 static int mdd_may_create(const struct lu_context *ctxt,
 187                           struct mdd_object *pobj, struct mdd_object *cobj)
 188 {
 189         ENTRY;
 190         if (cobj && lu_object_exists(&cobj->mod_obj.mo_lu))
 191                 RETURN(-EEXIST);
 192
 193         if (mdd_is_dead_obj(pobj))
 194                 RETURN(-ENOENT);
 195
 196         /*check pobj may create or not*/
 197         RETURN(0);
 198 }
 199
 200 static inline int __mdd_la_get(const struct lu_context *ctxt,
 201                                struct mdd_object *obj, struct lu_attr *la)
 202 {
 203         struct dt_object  *next = mdd_object_child(obj);
 204         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
 205         return next->do_ops->do_attr_get(ctxt, next, la);
 206 }
 207
 208 static void mdd_flags_xlate(struct mdd_object *obj, __u32 flags)
 209 {
 210         obj->mod_flags &= ~(APPEND_OBJ|IMMUTE_OBJ);
 211
 212         if (flags & LUSTRE_APPEND_FL)
 213                 obj->mod_flags |= APPEND_OBJ;
 214
 215         if (flags & LUSTRE_IMMUTABLE_FL)
 216                 obj->mod_flags |= IMMUTE_OBJ;
 217 }
 218
 219 static int mdd_get_flags(const struct lu_context *ctxt, struct mdd_object *obj)
 220 {
 221         struct lu_attr *la = &mdd_ctx_info(ctxt)->mti_la;
 222         int rc;
 223
 224         ENTRY;
 225         mdd_read_lock(ctxt, obj);
 226         rc = __mdd_la_get(ctxt, obj, la);
 227         mdd_read_unlock(ctxt, obj);
 228         if (rc == 0)
 229                 mdd_flags_xlate(obj, la->la_flags);
 230         RETURN(rc);
 231 }
 232
 233 /*Check whether it may delete the cobj under the pobj*/
 234 static int mdd_may_delete(const struct lu_context *ctxt,
 235                           struct mdd_object *pobj, struct mdd_object *cobj,
 236                           int is_dir)
 237 {
 238         struct mdd_device *mdd = mdo2mdd(&pobj->mod_obj);
 239         int rc = 0;
 240         ENTRY;
 241
 242         LASSERT(cobj && pobj);
 243
 244         if (!lu_object_exists(&cobj->mod_obj.mo_lu))
 245                 RETURN(-ENOENT);
 246
 247         if (mdd_is_immutable(cobj) || mdd_is_append(cobj))
 248                 RETURN(-EPERM);
 249
 250         if (is_dir) {
 251                 if (!S_ISDIR(mdd_object_type(cobj)))
 252                         RETURN(-ENOTDIR);
 253
 254                 if (lu_fid_eq(mdo2fid(cobj), &mdd->mdd_root_fid))
 255                         RETURN(-EBUSY);
 256
 257         } else if (S_ISDIR(mdd_object_type(cobj)))
 258                         RETURN(-EISDIR);
 259
 260         if (mdd_is_dead_obj(pobj))
 261                 RETURN(-ENOENT);
 262
 263         RETURN(rc);
 264 }
 265 /* get only inode attributes */
 266 static int __mdd_iattr_get(const struct lu_context *ctxt,
 267                            struct mdd_object *mdd_obj, struct md_attr *ma)
 268 {
 269         int rc = 0;
 270         ENTRY;
 271
 272         rc = __mdd_la_get(ctxt, mdd_obj, &ma->ma_attr);
 273         if (rc == 0)
 274                 ma->ma_valid = MA_INODE;
 275         RETURN(rc);
 276 }
 277 /* get lov EA only */
 278 static int __mdd_lmm_get(const struct lu_context *ctxt,
 279                          struct mdd_object *mdd_obj, struct md_attr *ma)
 280 {
 281         int rc;
 282
 283         LASSERT(ma->ma_lmm != NULL && ma->ma_lmm_size > 0);
 284         rc = mdd_get_md(ctxt, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size, 0,
 285                         MDS_LOV_MD_NAME);
 286         if (rc > 0) {
 287                 ma->ma_valid |= MA_LOV;
 288                 rc = 0;
 289         }
 290         RETURN(rc);
 291 }
 292
 293 /* get lmv EA only*/
 294 static int __mdd_lmv_get(const struct lu_context *ctxt,
 295                          struct mdd_object *mdd_obj, struct md_attr *ma)
 296 {
 297         int rc;
 298
 299         rc = mdd_get_md(ctxt, mdd_obj, ma->ma_lmv, &ma->ma_lmv_size, 0,
 300                         MDS_LMV_MD_NAME);
 301         if (rc > 0) {
 302                 ma->ma_valid |= MA_LMV;
 303                 rc = 0;
 304         }
 305         RETURN(rc);
 306 }
 307
 308 static int mdd_attr_get_internal(const struct lu_context *ctxt,
 309                                  struct mdd_object *mdd_obj,
 310                                  struct md_attr *ma)
 311 {
 312         int rc = 0;
 313         ENTRY;
 314
 315         if (ma->ma_need & MA_INODE)
 316                 rc = __mdd_iattr_get(ctxt, mdd_obj, ma);
 317
 318         if (rc == 0 && ma->ma_need & MA_LOV) {
 319                 if (S_ISREG(mdd_object_type(mdd_obj)) ||
 320                     S_ISDIR(mdd_object_type(mdd_obj)))
 321                         rc = __mdd_lmm_get(ctxt, mdd_obj, ma);
 322         }
 323         if (rc == 0 && ma->ma_need & MA_LMV) {
 324                 if (S_ISDIR(mdd_object_type(mdd_obj)))
 325                         rc = __mdd_lmv_get(ctxt, mdd_obj, ma);
 326         }
 327
 328         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
 329                         rc, ma->ma_valid);
 330         RETURN(rc);
 331 }
 332
 333 static inline int mdd_attr_get_internal_locked(const struct lu_context *ctxt,
 334                                                struct mdd_object *mdd_obj,
 335                                                struct md_attr *ma)
 336 {
 337         int rc;
 338         mdd_read_lock(ctxt, mdd_obj);
 339         rc = mdd_attr_get_internal(ctxt, mdd_obj, ma);
 340         mdd_read_unlock(ctxt, mdd_obj);
 341         return rc;
 342 }
 343
 344 static int mdd_attr_get(const struct lu_context *ctxt,
 345                         struct md_object *obj, struct md_attr *ma)
 346 {
 347         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 348         int                rc;
 349
 350         ENTRY;
 351         rc = mdd_attr_get_internal_locked(ctxt, mdd_obj, ma);
 352         RETURN(rc);
 353 }
 354
 355 static int mdd_xattr_get(const struct lu_context *ctxt, struct md_object *obj,
 356                          void *buf, int buf_len, const char *name)
 357 {
 358         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 359         struct dt_object  *next;
 360         int rc;
 361
 362         ENTRY;
 363
 364         LASSERT(lu_object_exists(&obj->mo_lu));
 365
 366         next = mdd_object_child(mdd_obj);
 367         mdd_read_lock(ctxt, mdd_obj);
 368         rc = next->do_ops->do_xattr_get(ctxt, next, buf, buf_len, name);
 369         mdd_read_unlock(ctxt, mdd_obj);
 370
 371         RETURN(rc);
 372 }
 373
 374 static int mdd_readlink(const struct lu_context *ctxt, struct md_object *obj,
 375                         void *buf, int buf_len)
 376 {
 377         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 378         struct dt_object  *next;
 379         loff_t             pos = 0;
 380         int                rc;
 381         ENTRY;
 382
 383         LASSERT(lu_object_exists(&obj->mo_lu));
 384
 385         next = mdd_object_child(mdd_obj);
 386         rc = next->do_body_ops->dbo_read(ctxt, next, buf, buf_len, &pos);
 387         RETURN(rc);
 388 }
 389 static int mdd_xattr_list(const struct lu_context *ctxt, struct md_object *obj,
 390                           void *buf, int buf_len)
 391 {
 392         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 393         struct dt_object  *next;
 394         int rc;
 395
 396         ENTRY;
 397
 398         LASSERT(lu_object_exists(&obj->mo_lu));
 399
 400         next = mdd_object_child(mdd_obj);
 401         rc = next->do_ops->do_xattr_list(ctxt, next, buf, buf_len);
 402
 403         RETURN(rc);
 404 }
 405
 406 enum mdd_txn_op {
 407         MDD_TXN_OBJECT_DESTROY_OP,
 408         MDD_TXN_OBJECT_CREATE_OP,
 409         MDD_TXN_ATTR_SET_OP,
 410         MDD_TXN_XATTR_SET_OP,
 411         MDD_TXN_INDEX_INSERT_OP,
 412         MDD_TXN_INDEX_DELETE_OP,
 413         MDD_TXN_LINK_OP,
 414         MDD_TXN_UNLINK_OP,
 415         MDD_TXN_RENAME_OP,
 416         MDD_TXN_CREATE_DATA_OP,
 417         MDD_TXN_MKDIR_OP
 418 };
 419
 420 struct mdd_txn_op_descr {
 421         enum mdd_txn_op mod_op;
 422         unsigned int    mod_credits;
 423 };
 424
 425 enum {
 426         MDD_TXN_OBJECT_DESTROY_CREDITS = 20,
 427         MDD_TXN_OBJECT_CREATE_CREDITS  = 20,
 428         MDD_TXN_ATTR_SET_CREDITS       = 20,
 429         MDD_TXN_XATTR_SET_CREDITS      = 20,
 430         MDD_TXN_INDEX_INSERT_CREDITS   = 20,
 431         MDD_TXN_INDEX_DELETE_CREDITS   = 20,
 432         MDD_TXN_LINK_CREDITS           = 20,
 433         MDD_TXN_UNLINK_CREDITS         = 20,
 434         MDD_TXN_RENAME_CREDITS         = 20,
 435         MDD_TXN_CREATE_DATA_CREDITS    = 20,
 436         MDD_TXN_MKDIR_CREDITS          = 20
 437 };
 438
 439 #define DEFINE_MDD_TXN_OP_DESC(opname)          \
 440 static const struct mdd_txn_op_descr opname = { \
 441         .mod_op      = opname ## _OP,           \
 442         .mod_credits = opname ## _CREDITS,      \
 443 }
 444
 445 /*
 446  * number of blocks to reserve for particular operations. Should be function
 447  * of ... something. Stub for now.
 448  */
 449 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_OBJECT_DESTROY);
 450 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_OBJECT_CREATE);
 451 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_ATTR_SET);
 452 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_XATTR_SET);
 453 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_INDEX_INSERT);
 454 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_INDEX_DELETE);
 455 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_LINK);
 456 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_UNLINK);
 457 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_RENAME);
 458 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_CREATE_DATA);
 459 DEFINE_MDD_TXN_OP_DESC(MDD_TXN_MKDIR);
 460
 461 static void mdd_txn_param_build(const struct lu_context *ctx,
 462                                 const struct mdd_txn_op_descr *opd)
 463 {
 464         mdd_ctx_info(ctx)->mti_param.tp_credits = opd->mod_credits;
 465 }
 466
 467 static int mdd_object_print(const struct lu_context *ctxt, void *cookie,
 468                             lu_printer_t p, const struct lu_object *o)
 469 {
 470         return (*p)(ctxt, cookie, LUSTRE_MDD0_NAME"-object@%p", o);
 471 }
 472
 473 static int mdd_mount(const struct lu_context *ctx, struct mdd_device *mdd)
 474 {
 475         int rc;
 476         struct dt_object *root;
 477         ENTRY;
 478
 479         root = dt_store_open(ctx, mdd->mdd_child, mdd_root_dir_name,
 480                              &mdd->mdd_root_fid);
 481         if (!IS_ERR(root)) {
 482                 LASSERT(root != NULL);
 483                 lu_object_put(ctx, &root->do_lu);
 484                 rc = 0;
 485         } else
 486                 rc = PTR_ERR(root);
 487
 488         RETURN(rc);
 489 }
 490
 491 static int mdd_device_init(const struct lu_context *ctx,
 492                            struct lu_device *d, struct lu_device *next)
 493 {
 494         struct mdd_device *mdd = lu2mdd_dev(d);
 495         int rc = 0;
 496         ENTRY;
 497
 498         mdd->mdd_child = lu2dt_dev(next);
 499
 500         RETURN(rc);
 501 }
 502
 503 static struct lu_device *mdd_device_fini(const struct lu_context *ctx,
 504                                          struct lu_device *d)
 505 {
 506         struct mdd_device *m = lu2mdd_dev(d);
 507         struct lu_device *next = &m->mdd_child->dd_lu_dev;
 508
 509         return next;
 510 }
 511 static void mdd_device_shutdown(const struct lu_context *ctxt,
 512                                 struct mdd_device *m)
 513 {
 514         mdd_fini_obd(ctxt, m);
 515 }
 516
 517 static int mdd_process_config(const struct lu_context *ctxt,
 518                               struct lu_device *d, struct lustre_cfg *cfg)
 519 {
 520         struct mdd_device *m    = lu2mdd_dev(d);
 521         struct dt_device  *dt   = m->mdd_child;
 522         struct lu_device  *next = &dt->dd_lu_dev;
 523         char              *dev = lustre_cfg_string(cfg, 0);
 524         int rc;
 525
 526         switch (cfg->lcfg_command) {
 527         case LCFG_SETUP:
 528                 rc = next->ld_ops->ldo_process_config(ctxt, next, cfg);
 529                 if (rc)
 530                         GOTO(out, rc);
 531                 dt->dd_ops->dt_conf_get(ctxt, dt, &m->mdd_dt_conf);
 532
 533                 rc = mdd_mount(ctxt, m);
 534                 if (rc)
 535                         GOTO(out, rc);
 536                 rc = mdd_init_obd(ctxt, m, dev);
 537                 if (rc) {
 538                         CERROR("lov init error %d \n", rc);
 539                         GOTO(out, rc);
 540                 }
 541                 break;
 542         case LCFG_CLEANUP:
 543                 mdd_device_shutdown(ctxt, m);
 544         default:
 545                 rc = next->ld_ops->ldo_process_config(ctxt, next, cfg);
 546                 break;
 547         }
 548 out:
 549         RETURN(rc);
 550 }
 551
 552 static int mdd_recovery_complete(const struct lu_context *ctxt,
 553                                  struct lu_device *d)
 554 {
 555         struct mdd_device *mdd = lu2mdd_dev(d);
 556         struct lu_device *next = &mdd->mdd_child->dd_lu_dev;
 557         int rc;
 558         ENTRY;
 559 /* TODO:
 560         rc = mdd_lov_set_nextid(ctx, mdd);
 561         if (rc) {
 562                 CERROR("%s: mdd_lov_set_nextid failed %d\n",
 563                        obd->obd_name, rc);
 564                 GOTO(out, rc);
 565         }
 566         rc = mdd_cleanup_unlink_llog(ctx, mdd);
 567
 568         obd_notify(obd->u.mds.mds_osc_obd, NULL,
 569                    obd->obd_async_recov ? OBD_NOTIFY_SYNC_NONBLOCK :
 570                    OBD_NOTIFY_SYNC, NULL);
 571 */
 572         /* TODO: orphans handling */
 573         rc = next->ld_ops->ldo_recovery_complete(ctxt, next);
 574         RETURN(rc);
 575 }
 576
 577 struct lu_device_operations mdd_lu_ops = {
 578         .ldo_object_alloc      = mdd_object_alloc,
 579         .ldo_process_config    = mdd_process_config,
 580         .ldo_recovery_complete = mdd_recovery_complete
 581 };
 582
 583 static struct lu_object_operations mdd_lu_obj_ops = {
 584         .loo_object_init    = mdd_object_init,
 585         .loo_object_start   = mdd_object_start,
 586         .loo_object_free    = mdd_object_free,
 587         .loo_object_print   = mdd_object_print
 588 };
 589
 590 void mdd_write_lock(const struct lu_context *ctxt, struct mdd_object *obj)
 591 {
 592         struct dt_object  *next = mdd_object_child(obj);
 593
 594         next->do_ops->do_write_lock(ctxt, next);
 595 }
 596
 597 void mdd_read_lock(const struct lu_context *ctxt, struct mdd_object *obj)
 598 {
 599         struct dt_object  *next = mdd_object_child(obj);
 600
 601         next->do_ops->do_read_lock(ctxt, next);
 602 }
 603
 604 void mdd_write_unlock(const struct lu_context *ctxt, struct mdd_object *obj)
 605 {
 606         struct dt_object  *next = mdd_object_child(obj);
 607
 608         next->do_ops->do_write_unlock(ctxt, next);
 609 }
 610
 611 void mdd_read_unlock(const struct lu_context *ctxt, struct mdd_object *obj)
 612 {
 613         struct dt_object  *next = mdd_object_child(obj);
 614
 615         next->do_ops->do_read_unlock(ctxt, next);
 616 }
 617
 618 static void mdd_lock2(const struct lu_context *ctxt,
 619                       struct mdd_object *o0, struct mdd_object *o1)
 620 {
 621         mdd_write_lock(ctxt, o0);
 622         mdd_write_lock(ctxt, o1);
 623 }
 624
 625 static void mdd_unlock2(const struct lu_context *ctxt,
 626                         struct mdd_object *o0, struct mdd_object *o1)
 627 {
 628         mdd_write_unlock(ctxt, o1);
 629         mdd_write_unlock(ctxt, o0);
 630 }
 631
 632 static struct thandle* mdd_trans_start(const struct lu_context *ctxt,
 633                                        struct mdd_device *mdd)
 634 {
 635         struct txn_param *p = &mdd_ctx_info(ctxt)->mti_param;
 636
 637         return mdd_child_ops(mdd)->dt_trans_start(ctxt, mdd->mdd_child, p);
 638 }
 639
 640 static void mdd_trans_stop(const struct lu_context *ctxt,
 641                            struct mdd_device *mdd, int result,
 642                            struct thandle *handle)
 643 {
 644         handle->th_result = result;
 645         mdd_child_ops(mdd)->dt_trans_stop(ctxt, handle);
 646 }
 647
 648 static int __mdd_object_create(const struct lu_context *ctxt,
 649                                struct mdd_object *obj, struct md_attr *ma,
 650                                struct thandle *handle)
 651 {
 652         struct dt_object *next;
 653         struct lu_attr *attr = &ma->ma_attr;
 654         int rc;
 655         ENTRY;
 656
 657         if (!lu_object_exists(mdd2lu_obj(obj))) {
 658                 next = mdd_object_child(obj);
 659                 rc = next->do_ops->do_create(ctxt, next, attr, handle);
 660         } else
 661                 rc = -EEXIST;
 662
 663         LASSERT(ergo(rc == 0, lu_object_exists(mdd2lu_obj(obj))));
 664
 665         RETURN(rc);
 666 }
 667
 668 int mdd_attr_set_internal(const struct lu_context *ctxt, struct mdd_object *o,
 669                           const struct lu_attr *attr, struct thandle *handle)
 670 {
 671         struct dt_object *next;
 672
 673         LASSERT(lu_object_exists(mdd2lu_obj(o)));
 674         next = mdd_object_child(o);
 675         return next->do_ops->do_attr_set(ctxt, next, attr, handle);
 676 }
 677
 678 int mdd_attr_set_internal_locked(const struct lu_context *ctxt,
 679                                  struct mdd_object *o,
 680                                  const struct lu_attr *attr,
 681                                  struct thandle *handle)
 682 {
 683         int rc;
 684         mdd_write_lock(ctxt, o);
 685         rc = mdd_attr_set_internal(ctxt, o, attr, handle);
 686         mdd_write_unlock(ctxt, o);
 687         return rc;
 688 }
 689
 690 static int __mdd_xattr_set(const struct lu_context *ctxt, struct mdd_object *o,
 691                            const void *buf, int buf_len, const char *name,
 692                            int fl, struct thandle *handle)
 693 {
 694         struct dt_object *next;
 695         int rc = 0;
 696         ENTRY;
 697
 698         LASSERT(lu_object_exists(mdd2lu_obj(o)));
 699         next = mdd_object_child(o);
 700         if (buf && buf_len > 0) {
 701                 rc = next->do_ops->do_xattr_set(ctxt, next, buf, buf_len, name,
 702                                                 0, handle);
 703         }else if (buf == NULL && buf_len == 0) {
 704                 rc = next->do_ops->do_xattr_del(ctxt, next, name, handle);
 705         }
 706         RETURN(rc);
 707 }
 708 /* this gives the same functionality as the code between
 709  * sys_chmod and inode_setattr
 710  * chown_common and inode_setattr
 711  * utimes and inode_setattr
 712  * This API is ported from mds_fix_attr but remove some unnecesssary stuff.
 713  * and port to
 714  */
 715 int mdd_fix_attr(const struct lu_context *ctxt, struct mdd_object *obj,
 716                  const struct md_attr *ma, struct lu_attr *la)
 717 {
 718         struct lu_attr   *tmp_la = &mdd_ctx_info(ctxt)->mti_la;
 719         time_t            now = CURRENT_SECONDS;
 720         int               rc;
 721         ENTRY;
 722
 723         rc = __mdd_la_get(ctxt, obj, tmp_la);
 724         if (rc)
 725                 RETURN(rc);
 726         /*XXX Check permission */
 727         if (mdd_is_immutable(obj) || mdd_is_append(obj)) {
 728
 729                 /*If only change flags of the object, we should
 730                  * let it pass, but also need capability check
 731                  * here if (!capable(CAP_LINUX_IMMUTABLE)),
 732                  * fix it, when implement capable in mds*/
 733                 if (la->la_valid & ~LA_FLAGS)
 734                         RETURN(-EPERM);
 735
 736                 /*According to Ext3 implementation on this, the
 737                  *Ctime will be changed, but not clear why?*/
 738                 la->la_ctime = now;
 739                 la->la_valid |= LA_CTIME;
 740                 RETURN(0);
 741         }
 742         if (!(la->la_valid & LA_CTIME)) {
 743                 la->la_ctime = now;
 744                 la->la_valid |= LA_CTIME;
 745         } else
 746                 /*According to original MDS implementation, it
 747                  * will clear ATTR_CTIME_SET flags, but seems
 748                  * no sense, should clear ATTR_CTIME? XXX*/
 749                 la->la_valid &= ~LA_CTIME;
 750
 751         if (!(la->la_valid & LA_ATIME)) {
 752                 la->la_atime = now;
 753                 la->la_valid |= LA_ATIME;
 754         }
 755         if (!(la->la_valid & LA_MTIME)) {
 756                 la->la_mtime = now;
 757                 la->la_valid |= LA_MTIME;
 758         }
 759
 760
 761 #if 0
 762         /* times */
 763         if ((ia_valid & (ATTR_MTIME|ATTR_ATIME)) == (ATTR_MTIME|ATTR_ATIME)) {
 764                 if (current->fsuid != inode->i_uid &&
 765                     (error = ll_permission(inode, MAY_WRITE, NULL)) != 0)
 766                         RETURN(error);
 767         }
 768         if (ia_valid & ATTR_SIZE &&
 769             /* NFSD hack for open(O_CREAT|O_TRUNC)=mknod+truncate (bug 5781) */
 770             !(rec->ur_uc.luc_fsuid == inode->i_uid &&
 771               ia_valid & MDS_OPEN_OWNEROVERRIDE)) {
 772                 if ((error = ll_permission(inode, MAY_WRITE, NULL)) != 0)
 773                         RETURN(error);
 774         }
 775 #endif
 776
 777         if (la->la_valid & (LA_UID | LA_GID)) {
 778                 /* chown */
 779
 780                 if (mdd_is_immutable(obj) || mdd_is_append(obj))
 781                         RETURN(-EPERM);
 782                 if (la->la_uid == (uid_t) -1)
 783                         la->la_uid = tmp_la->la_uid;
 784                 if (la->la_gid == (gid_t) -1)
 785                         la->la_gid = tmp_la->la_gid;
 786                 if (!(la->la_valid & LA_MODE))
 787                         la->la_mode = tmp_la->la_mode;
 788                 /*
 789                  * If the user or group of a non-directory has been
 790                  * changed by a non-root user, remove the setuid bit.
 791                  * 19981026 David C Niemi <niemi@tux.org>
 792                  *
 793                  * Changed this to apply to all users, including root,
 794                  * to avoid some races. This is the behavior we had in
 795                  * 2.0. The check for non-root was definitely wrong
 796                  * for 2.2 anyway, as it should have been using
 797                  * CAP_FSETID rather than fsuid -- 19990830 SD.
 798                  */
 799                 if ((tmp_la->la_mode & S_ISUID) == S_ISUID &&
 800                     !S_ISDIR(tmp_la->la_mode)) {
 801                         la->la_mode &= ~S_ISUID;
 802                         la->la_valid |= LA_MODE;
 803                 }
 804                 /*
 805                  * Likewise, if the user or group of a non-directory
 806                  * has been changed by a non-root user, remove the
 807                  * setgid bit UNLESS there is no group execute bit
 808                  * (this would be a file marked for mandatory
 809                  * locking).  19981026 David C Niemi <niemi@tux.org>
 810                  *
 811                  * Removed the fsuid check (see the comment above) --
 812                  * 19990830 SD.
 813                  */
 814                 if (((tmp_la->la_mode & (S_ISGID | S_IXGRP)) ==
 815                      (S_ISGID | S_IXGRP)) && !S_ISDIR(tmp_la->la_mode)) {
 816                         la->la_mode &= ~S_ISGID;
 817                         la->la_valid |= LA_MODE;
 818                 }
 819         } else if (la->la_valid & LA_MODE) {
 820                 int mode = la->la_mode;
 821                 /* chmod */
 822                 if (la->la_mode == (umode_t)-1)
 823                         mode = tmp_la->la_mode;
 824                 la->la_mode =
 825                         (mode & S_IALLUGO) | (tmp_la->la_mode & ~S_IALLUGO);
 826         }
 827         RETURN(rc);
 828 }
 829
 830
 831 /* set attr and LOV EA at once, return updated attr */
 832 static int mdd_attr_set(const struct lu_context *ctxt,
 833                         struct md_object *obj,
 834                         const struct md_attr *ma)
 835 {
 836         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 837         struct mdd_device *mdd = mdo2mdd(obj);
 838         struct thandle *handle;
 839         struct lov_mds_md *lmm = NULL;
 840         int  rc = 0, lmm_size = 0, max_size;
 841         struct lu_attr *la_copy = &mdd_ctx_info(ctxt)->mti_la_for_fix;
 842         ENTRY;
 843
 844         mdd_txn_param_build(ctxt, &MDD_TXN_ATTR_SET);
 845         handle = mdd_trans_start(ctxt, mdd);
 846         if (IS_ERR(handle))
 847                 RETURN(PTR_ERR(handle));
 848         /*TODO: add lock here*/
 849         /* start a log jounal handle if needed */
 850         if (S_ISREG(mdd_object_type(mdd_obj)) &&
 851             ma->ma_attr.la_valid & (LA_UID | LA_GID)) {
 852                 max_size = mdd_lov_mdsize(ctxt, mdd);
 853                 OBD_ALLOC(lmm, max_size);
 854                 if (lmm == NULL)
 855                         GOTO(cleanup, rc = -ENOMEM);
 856
 857                 rc = mdd_get_md(ctxt, mdd_obj, lmm, &lmm_size, 1,
 858                                 MDS_LOV_MD_NAME);
 859
 860                 if (rc < 0)
 861                         GOTO(cleanup, rc);
 862         }
 863
 864         if (ma->ma_attr.la_valid & (ATTR_MTIME | ATTR_CTIME))
 865                 CDEBUG(D_INODE, "setting mtime "LPU64", ctime "LPU64"\n",
 866                        ma->ma_attr.la_mtime, ma->ma_attr.la_ctime);
 867
 868         *la_copy = ma->ma_attr;
 869         mdd_write_lock(ctxt, mdd_obj);
 870         rc = mdd_fix_attr(ctxt, mdd_obj, ma, la_copy);
 871         mdd_write_unlock(ctxt, mdd_obj);
 872         if (rc)
 873                 GOTO(cleanup, rc);
 874
 875         if (la_copy->la_valid & LA_FLAGS) {
 876                 rc = mdd_attr_set_internal_locked(ctxt, mdd_obj, la_copy,
 877                                                   handle);
 878                 if (rc == 0)
 879                         mdd_flags_xlate(mdd_obj, la_copy->la_flags);
 880         } else if (la_copy->la_valid) {            /* setattr */
 881                 rc = mdd_attr_set_internal_locked(ctxt, mdd_obj, la_copy,
 882                                                   handle);
 883                 /* journal chown/chgrp in llog, just like unlink */
 884                 if (rc == 0 && lmm_size){
 885                         /*TODO set_attr llog */
 886                 }
 887         }
 888
 889         if (rc == 0 && ma->ma_valid & MA_LOV) {
 890                 umode_t mode;
 891
 892                 mode = mdd_object_type(mdd_obj);
 893                 if (S_ISREG(mode) || S_ISDIR(mode)) {
 894                         /*TODO check permission*/
 895                         rc = mdd_lov_set_md(ctxt, NULL, mdd_obj, ma->ma_lmm,
 896                                             ma->ma_lmm_size, handle, 1);
 897                 }
 898
 899         }
 900 cleanup:
 901         mdd_trans_stop(ctxt, mdd, rc, handle);
 902         if (rc == 0 && lmm_size) {
 903                 /*set obd attr, if needed*/
 904                 rc = mdd_lov_setattr_async(ctxt, mdd_obj, lmm, lmm_size);
 905         }
 906         if (lmm != NULL) {
 907                 OBD_FREE(lmm, max_size);
 908         }
 909
 910         RETURN(rc);
 911 }
 912
 913 int mdd_xattr_set_txn(const struct lu_context *ctxt, struct mdd_object *obj,
 914                       const void *buf, int buf_len, const char *name, int fl,
 915                       struct thandle *handle)
 916 {
 917         int  rc;
 918         ENTRY;
 919
 920         mdd_write_lock(ctxt, obj);
 921         rc = __mdd_xattr_set(ctxt, obj, buf, buf_len, name, fl, handle);
 922         mdd_write_unlock(ctxt, obj);
 923
 924         RETURN(rc);
 925 }
 926
 927 static int mdd_xattr_set(const struct lu_context *ctxt, struct md_object *obj,
 928                          const void *buf, int buf_len, const char *name,
 929                          int fl)
 930 {
 931         struct mdd_device *mdd = mdo2mdd(obj);
 932         struct thandle *handle;
 933         int  rc;
 934         ENTRY;
 935
 936         mdd_txn_param_build(ctxt, &MDD_TXN_XATTR_SET);
 937         handle = mdd_trans_start(ctxt, mdd);
 938         if (IS_ERR(handle))
 939                 RETURN(PTR_ERR(handle));
 940
 941         rc = mdd_xattr_set_txn(ctxt, md2mdd_obj(obj), buf, buf_len, name,
 942                                fl, handle);
 943
 944         mdd_trans_stop(ctxt, mdd, rc, handle);
 945
 946         RETURN(rc);
 947 }
 948
 949 static int __mdd_xattr_del(const struct lu_context *ctxt,struct mdd_device *mdd,
 950                            struct mdd_object *obj,
 951                            const char *name, struct thandle *handle)
 952 {
 953         struct dt_object *next;
 954
 955         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
 956         next = mdd_object_child(obj);
 957         return next->do_ops->do_xattr_del(ctxt, next, name, handle);
 958 }
 959
 960 int mdd_xattr_del(const struct lu_context *ctxt, struct md_object *obj,
 961                   const char *name)
 962 {
 963         struct mdd_object *mdd_obj = md2mdd_obj(obj);
 964         struct mdd_device *mdd = mdo2mdd(obj);
 965         struct thandle *handle;
 966         int  rc;
 967         ENTRY;
 968
 969         mdd_txn_param_build(ctxt, &MDD_TXN_XATTR_SET);
 970         handle = mdd_trans_start(ctxt, mdd);
 971         if (IS_ERR(handle))
 972                 RETURN(PTR_ERR(handle));
 973
 974         mdd_write_lock(ctxt, mdd_obj);
 975         rc = __mdd_xattr_del(ctxt, mdd, md2mdd_obj(obj), name, handle);
 976         mdd_write_unlock(ctxt, mdd_obj);
 977
 978         mdd_trans_stop(ctxt, mdd, rc, handle);
 979
 980         RETURN(rc);
 981 }
 982
 983 static int __mdd_index_insert_only(const struct lu_context *ctxt,
 984                                    struct mdd_object *pobj,
 985                                    const struct lu_fid *lf,
 986                                    const char *name, struct thandle *th)
 987 {
 988         int rc;
 989         struct dt_object *next = mdd_object_child(pobj);
 990         ENTRY;
 991
 992         if (dt_try_as_dir(ctxt, next))
 993                 rc = next->do_index_ops->dio_insert(ctxt, next,
 994                                          (struct dt_rec *)lf,
 995                                          (struct dt_key *)name, th);
 996         else
 997                 rc = -ENOTDIR;
 998         RETURN(rc);
 999 }
1000
1001 /* insert new index, add reference if isdir, update times */
1002 static int __mdd_index_insert(const struct lu_context *ctxt,
1003                              struct mdd_object *pobj, const struct lu_fid *lf,
1004                              const char *name, int isdir, struct thandle *th)
1005 {
1006         int rc;
1007         struct dt_object *next = mdd_object_child(pobj);
1008         ENTRY;
1009
1010 #if 0
1011         struct lu_attr   *la = &mdd_ctx_info(ctxt)->mti_la;
1012 #endif
1013
1014         if (dt_try_as_dir(ctxt, next))
1015                 rc = next->do_index_ops->dio_insert(ctxt, next,
1016                                          (struct dt_rec *)lf,
1017                                          (struct dt_key *)name, th);
1018         else
1019                 rc = -ENOTDIR;
1020
1021         if (rc == 0) {
1022                 if (isdir)
1023                         __mdd_ref_add(ctxt, pobj, th);
1024 #if 0
1025                 la->la_valid = LA_MTIME|LA_CTIME;
1026                 la->la_atime = ma->ma_attr.la_atime;
1027                 la->la_ctime = ma->ma_attr.la_ctime;
1028                 rc = mdd_attr_set_internal(ctxt, mdd_obj, la, handle);
1029 #endif
1030         }
1031         return rc;
1032 }
1033
1034 static int __mdd_index_delete(const struct lu_context *ctxt,
1035                               struct mdd_object *pobj, const char *name,
1036                               struct thandle *handle)
1037 {
1038         int rc;
1039         struct dt_object *next = mdd_object_child(pobj);
1040         ENTRY;
1041
1042         if (dt_try_as_dir(ctxt, next))
1043                 rc = next->do_index_ops->dio_delete(ctxt, next,
1044                                         (struct dt_key *)name, handle);
1045         else
1046                 rc = -ENOTDIR;
1047         RETURN(rc);
1048 }
1049
1050 static int mdd_link_sanity_check(const struct lu_context *ctxt,
1051                                  struct mdd_object *tgt_obj,
1052                                  struct mdd_object *src_obj)
1053 {
1054         int rc;
1055
1056         rc = mdd_may_create(ctxt, tgt_obj, NULL);
1057         if (rc)
1058                 RETURN(rc);
1059         if (S_ISDIR(mdd_object_type(src_obj)))
1060                 RETURN(-EPERM);
1061
1062         if (mdd_is_immutable(src_obj) || mdd_is_append(src_obj))
1063                 RETURN(-EPERM);
1064
1065         RETURN(rc);
1066 }
1067
1068 static int mdd_link(const struct lu_context *ctxt, struct md_object *tgt_obj,
1069                     struct md_object *src_obj, const char *name)
1070 {
1071         struct mdd_object *mdd_tobj = md2mdd_obj(tgt_obj);
1072         struct mdd_object *mdd_sobj = md2mdd_obj(src_obj);
1073         struct mdd_device *mdd = mdo2mdd(src_obj);
1074         struct thandle *handle;
1075         int rc;
1076         ENTRY;
1077
1078         mdd_txn_param_build(ctxt, &MDD_TXN_LINK);
1079         handle = mdd_trans_start(ctxt, mdd);
1080         if (IS_ERR(handle))
1081                 RETURN(PTR_ERR(handle));
1082
1083         mdd_lock2(ctxt, mdd_tobj, mdd_sobj);
1084
1085         rc = mdd_link_sanity_check(ctxt, mdd_tobj, mdd_sobj);
1086         if (rc)
1087                 GOTO(out, rc);
1088
1089         rc = __mdd_index_insert_only(ctxt, mdd_tobj, mdo2fid(mdd_sobj),
1090                                      name, handle);
1091         if (rc == 0)
1092                 __mdd_ref_add(ctxt, mdd_sobj, handle);
1093
1094 out:
1095         mdd_unlock2(ctxt, mdd_tobj, mdd_sobj);
1096         mdd_trans_stop(ctxt, mdd, rc, handle);
1097         RETURN(rc);
1098 }
1099
1100 /*
1101  * Check that @dir contains no entries except (possibly) dot and dotdot.
1102  *
1103  * Returns:
1104  *
1105  *             0        empty
1106  *    -ENOTEMPTY        not empty
1107  *           -ve        other error
1108  *
1109  */
1110 static int mdd_dir_is_empty(const struct lu_context *ctx,
1111                             struct mdd_object *dir)
1112 {
1113         struct dt_it     *it;
1114         struct dt_object *obj;
1115         struct dt_it_ops *iops;
1116         int result;
1117
1118         obj = mdd_object_child(dir);
1119         iops = &obj->do_index_ops->dio_it;
1120         it = iops->init(ctx, obj);
1121         if (it != NULL) {
1122                 result = iops->get(ctx, it, (const void *)"");
1123                 if (result > 0) {
1124                         int i;
1125                         for (result = 0, i = 0; result == 0 && i < 3; ++i)
1126                                 result = iops->next(ctx, it);
1127                         iops->put(ctx, it);
1128                         if (result == 0)
1129                                 result = -ENOTEMPTY;
1130                         else if (result == +1)
1131                                 result = 0;
1132                 } else if (result == 0)
1133                         /*
1134                          * Huh? Index contains no zero key?
1135                          */
1136                         result = -EIO;
1137                 iops->fini(ctx, it);
1138         } else
1139                 result = -ENOMEM;
1140         return result;
1141 }
1142
1143 /* return md_attr back,
1144  * if it is last unlink then return lov ea + llog cookie*/
1145 static int __mdd_finish_unlink(const struct lu_context *ctxt,
1146                                struct mdd_object *obj, struct md_attr *ma)
1147 {
1148         int rc;
1149         ENTRY;
1150
1151         rc = __mdd_iattr_get(ctxt, obj, ma);
1152         if (rc == 0 && ma->ma_attr.la_nlink == 0) {
1153                 if (atomic_read(&obj->mod_count) == 0) {
1154                         mdd_set_dead_obj(obj);
1155                         if (S_ISREG(mdd_object_type(obj))) {
1156                                 rc = __mdd_lmm_get(ctxt, obj, ma);
1157                                 if (ma->ma_valid & MA_LOV)
1158                                         rc = mdd_unlink_log(ctxt,
1159                                                     mdo2mdd(&obj->mod_obj),
1160                                                     obj, ma);
1161                         }
1162                 } else {
1163                         /* add new orphan */
1164                         //__mdd_orphan_add(ctxt, obj);
1165                 }
1166         }
1167         RETURN(rc);
1168 }
1169
1170 static int mdd_unlink_sanity_check(const struct lu_context *ctxt,
1171                                    struct mdd_object *pobj,
1172                                    struct mdd_object *cobj,
1173                                    struct md_attr *ma)
1174 {
1175         struct dt_object  *dt_cobj  = mdd_object_child(cobj);
1176         int rc = 0;
1177         ENTRY;
1178
1179         rc = mdd_may_delete(ctxt, pobj, cobj, S_ISDIR(ma->ma_attr.la_mode));
1180         if (rc)
1181                 RETURN(rc);
1182
1183         if (S_ISDIR(mdd_object_type(cobj)) &&
1184             dt_try_as_dir(ctxt, dt_cobj)) {
1185                 rc = mdd_dir_is_empty(ctxt, cobj);
1186                 if (rc != 0)
1187                         RETURN(rc);
1188         }
1189
1190         RETURN(rc);
1191 }
1192
1193 static int mdd_unlink(const struct lu_context *ctxt, struct md_object *pobj,
1194                       struct md_object *cobj, const char *name,
1195                       struct md_attr *ma)
1196 {
1197         struct mdd_device *mdd = mdo2mdd(pobj);
1198         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1199         struct mdd_object *mdd_cobj = md2mdd_obj(cobj);
1200         struct thandle    *handle;
1201         int rc;
1202         ENTRY;
1203
1204         mdd_txn_param_build(ctxt, &MDD_TXN_UNLINK);
1205         handle = mdd_trans_start(ctxt, mdd);
1206         if (IS_ERR(handle))
1207                 RETURN(PTR_ERR(handle));
1208
1209         mdd_lock2(ctxt, mdd_pobj, mdd_cobj);
1210
1211         rc = mdd_unlink_sanity_check(ctxt, mdd_pobj, mdd_cobj, ma);
1212         if (rc)
1213                 GOTO(cleanup, rc);
1214
1215
1216         rc = __mdd_index_delete(ctxt, mdd_pobj, name, handle);
1217         if (rc)
1218                 GOTO(cleanup, rc);
1219
1220         __mdd_ref_del(ctxt, mdd_cobj, handle);
1221         if (S_ISDIR(lu_object_attr(&cobj->mo_lu))) {
1222                 /* unlink dot */
1223                 __mdd_ref_del(ctxt, mdd_cobj, handle);
1224                 /* unlink dotdot */
1225                 __mdd_ref_del(ctxt, mdd_pobj, handle);
1226         }
1227
1228         rc = __mdd_finish_unlink(ctxt, mdd_cobj, ma);
1229
1230 cleanup:
1231         mdd_unlock2(ctxt, mdd_pobj, mdd_cobj);
1232         mdd_trans_stop(ctxt, mdd, rc, handle);
1233         RETURN(rc);
1234 }
1235
1236 static int mdd_parent_fid(const struct lu_context *ctxt,
1237                           struct mdd_object *obj,
1238                           struct lu_fid *fid)
1239 {
1240         return mdd_lookup(ctxt, &obj->mod_obj, dotdot, fid);
1241 }
1242
1243 /*
1244  * return 0: if p2 is the parent of p1
1245  * otherwise: other_value
1246  */
1247 static int mdd_is_parent(const struct lu_context *ctxt,
1248                          struct mdd_device *mdd,
1249                          struct mdd_object *p1,
1250                          struct mdd_object *p2)
1251 {
1252         struct lu_fid * pfid;
1253         struct mdd_object *parent = NULL;
1254         int rc;
1255         ENTRY;
1256
1257         pfid = &mdd_ctx_info(ctxt)->mti_fid;
1258         if (lu_fid_eq(mdo2fid(p1), &mdd->mdd_root_fid))
1259                 RETURN(1);
1260         for(;;) {
1261                 rc = mdd_parent_fid(ctxt, p1, pfid);
1262                 if (rc)
1263                         GOTO(out, rc);
1264                 if (lu_fid_eq(pfid, mdo2fid(p2)))
1265                         GOTO(out, rc = 0);
1266                 if (lu_fid_eq(pfid, &mdd->mdd_root_fid))
1267                         GOTO(out, rc = 1);
1268                 if (parent)
1269                         mdd_object_put(ctxt, parent);
1270                 parent = mdd_object_find(ctxt, mdd, pfid);
1271                 if (IS_ERR(parent))
1272                         GOTO(out, rc = PTR_ERR(parent));
1273                 p1 = parent;
1274         }
1275 out:
1276         if (parent && !IS_ERR(parent))
1277                 mdd_object_put(ctxt, parent);
1278         RETURN(rc);
1279 }
1280
1281 static int mdd_rename_lock(const struct lu_context *ctxt,
1282                            struct mdd_device *mdd,
1283                            struct mdd_object *src_pobj,
1284                            struct mdd_object *tgt_pobj)
1285 {
1286         ENTRY;
1287
1288         if (src_pobj == tgt_pobj) {
1289                 mdd_write_lock(ctxt, src_pobj);
1290                 RETURN(0);
1291         }
1292         /*compared the parent child relationship of src_p&tgt_p*/
1293         if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(src_pobj))){
1294                 mdd_lock2(ctxt, src_pobj, tgt_pobj);
1295                 RETURN(0);
1296         } else if (lu_fid_eq(&mdd->mdd_root_fid, mdo2fid(tgt_pobj))) {
1297                 mdd_lock2(ctxt, tgt_pobj, src_pobj);
1298                 RETURN(0);
1299         }
1300
1301         if (!mdd_is_parent(ctxt, mdd, src_pobj, tgt_pobj)) {
1302                 mdd_lock2(ctxt, tgt_pobj, src_pobj);
1303                 RETURN(0);
1304         }
1305
1306         mdd_lock2(ctxt, src_pobj, tgt_pobj);
1307
1308         RETURN(0);
1309 }
1310
1311 static void mdd_rename_unlock(const struct lu_context *ctxt,
1312                               struct mdd_object *src_pobj,
1313                               struct mdd_object *tgt_pobj)
1314 {
1315         mdd_write_unlock(ctxt, src_pobj);
1316         if (src_pobj != tgt_pobj)
1317                 mdd_write_unlock(ctxt, tgt_pobj);
1318 }
1319
1320 static int mdd_rename_sanity_check(const struct lu_context *ctxt,
1321                                    struct mdd_object *src_pobj,
1322                                    struct mdd_object *tgt_pobj,
1323                                    struct mdd_object *sobj,
1324                                    struct mdd_object *tobj)
1325 {
1326         struct mdd_device *mdd =mdo2mdd(&src_pobj->mod_obj);
1327         int rc = 0, src_is_dir, tgt_is_dir;
1328         ENTRY;
1329
1330         src_is_dir = S_ISDIR(mdd_object_type(sobj));
1331         rc = mdd_may_delete(ctxt, src_pobj, sobj, src_is_dir);
1332         if (rc)
1333                 GOTO(out, rc);
1334
1335         if (!tobj) {
1336                 rc = mdd_may_create(ctxt, tgt_pobj, NULL);
1337                 if (rc)
1338                         GOTO(out, rc);
1339                 if (!mdd_is_parent(ctxt, mdd, tgt_pobj, sobj))
1340                         GOTO(out, rc = -EINVAL);
1341                 GOTO(out, rc);
1342         }
1343
1344         /* source should not be ancestor of target */
1345         if (!mdd_is_parent(ctxt, mdd, tobj, sobj))
1346                 GOTO(out, rc = -EINVAL);
1347
1348         rc = mdd_may_delete(ctxt, tgt_pobj, tobj, src_is_dir);
1349         if (rc)
1350                 GOTO(out, rc);
1351
1352         tgt_is_dir = S_ISDIR(mdd_object_type(tobj));
1353         if (tgt_is_dir && mdd_dir_is_empty(ctxt, tobj))
1354                 GOTO(out, rc = -ENOTEMPTY);
1355 out:
1356         mdd_object_put(ctxt, sobj);
1357         RETURN(rc);
1358 }
1359
1360 static int mdd_rename(const struct lu_context *ctxt, struct md_object *src_pobj,
1361                       struct md_object *tgt_pobj, const struct lu_fid *lf,
1362                       const char *sname, struct md_object *tobj,
1363                       const char *tname, struct md_attr *ma)
1364 {
1365         struct mdd_device *mdd = mdo2mdd(src_pobj);
1366         struct mdd_object *mdd_spobj = md2mdd_obj(src_pobj);
1367         struct mdd_object *mdd_tpobj = md2mdd_obj(tgt_pobj);
1368         struct mdd_object *mdd_sobj = mdd_object_find(ctxt, mdd, lf);
1369         struct mdd_object *mdd_tobj = NULL;
1370         struct thandle *handle;
1371         int is_dir = S_ISDIR(mdd_object_type(mdd_sobj));
1372         int rc;
1373         ENTRY;
1374
1375         if (tobj)
1376                 mdd_tobj = md2mdd_obj(tobj);
1377
1378         /*XXX: shouldn't this check be done under lock below? */
1379         rc = mdd_rename_sanity_check(ctxt, mdd_spobj, mdd_tpobj,
1380                                      mdd_sobj, mdd_tobj);
1381         if (rc)
1382                 RETURN(rc);
1383
1384         mdd_txn_param_build(ctxt, &MDD_TXN_RENAME);
1385         handle = mdd_trans_start(ctxt, mdd);
1386         if (IS_ERR(handle))
1387                 RETURN(PTR_ERR(handle));
1388
1389         /*FIXME: Should consider tobj and sobj too in rename_lock*/
1390         rc = mdd_rename_lock(ctxt, mdd, mdd_spobj, mdd_tpobj);
1391         if (rc)
1392                 GOTO(cleanup_unlocked, rc);
1393
1394         rc = __mdd_index_delete(ctxt, mdd_spobj, sname, handle);
1395         if (rc)
1396                 GOTO(cleanup, rc);
1397
1398         /*if sobj is dir, its parent object nlink should be dec too*/
1399         if (is_dir)
1400                 __mdd_ref_del(ctxt, mdd_spobj, handle);
1401
1402         if (tobj) {
1403                 rc = __mdd_index_delete(ctxt, mdd_tpobj, tname, handle);
1404                 if (rc)
1405                         GOTO(cleanup, rc);
1406         }
1407
1408         rc = __mdd_index_insert(ctxt, mdd_tpobj, lf, tname, is_dir, handle);
1409         if (rc)
1410                 GOTO(cleanup, rc);
1411
1412         if (tobj && lu_object_exists(&tobj->mo_lu)) {
1413                 mdd_write_lock(ctxt, mdd_tobj);
1414                 __mdd_ref_del(ctxt, mdd_tobj, handle);
1415                 /* remove dot reference */
1416                 if (is_dir)
1417                         __mdd_ref_del(ctxt, mdd_tobj, handle);
1418
1419                 rc = __mdd_finish_unlink(ctxt, mdd_tobj, ma);
1420                 mdd_write_unlock(ctxt, mdd_tobj);
1421         }
1422 cleanup:
1423         mdd_rename_unlock(ctxt, mdd_spobj, mdd_tpobj);
1424 cleanup_unlocked:
1425         mdd_trans_stop(ctxt, mdd, rc, handle);
1426         RETURN(rc);
1427 }
1428
1429 static int mdd_lookup(const struct lu_context *ctxt, struct md_object *pobj,
1430                       const char *name, struct lu_fid* fid)
1431 {
1432         struct mdd_object   *mdd_obj = md2mdd_obj(pobj);
1433         struct dt_object    *dir = mdd_object_child(mdd_obj);
1434         struct dt_rec       *rec    = (struct dt_rec *)fid;
1435         const struct dt_key *key = (const struct dt_key *)name;
1436         int rc;
1437         ENTRY;
1438
1439         if (mdd_is_dead_obj(mdd_obj))
1440                 RETURN(-ESTALE);
1441         mdd_read_lock(ctxt, mdd_obj);
1442         if (S_ISDIR(mdd_object_type(mdd_obj)) && dt_try_as_dir(ctxt, dir))
1443                 rc = dir->do_index_ops->dio_lookup(ctxt, dir, rec, key);
1444         else
1445                 rc = -ENOTDIR;
1446         mdd_read_unlock(ctxt, mdd_obj);
1447         RETURN(rc);
1448 }
1449
1450 static int __mdd_object_initialize(const struct lu_context *ctxt,
1451                                    const struct lu_fid *pfid,
1452                                    struct mdd_object *child,
1453                                    struct md_attr *ma, struct thandle *handle)
1454 {
1455         int rc;
1456         ENTRY;
1457
1458         /* update attributes for child.
1459          * FIXME:
1460          *  (1) the valid bits should be converted between Lustre and Linux;
1461          *  (2) maybe, the child attributes should be set in OSD when creation.
1462          */
1463
1464         rc = mdd_attr_set_internal(ctxt, child, &ma->ma_attr, handle);
1465         if (rc != 0)
1466                 RETURN(rc);
1467
1468         if (S_ISDIR(ma->ma_attr.la_mode)) {
1469                 /* add . and .. for newly created dir */
1470                 __mdd_ref_add(ctxt, child, handle);
1471                 rc = __mdd_index_insert_only(ctxt, child, mdo2fid(child),
1472                                              dot, handle);
1473                 if (rc == 0) {
1474                         rc = __mdd_index_insert_only(ctxt, child, pfid,
1475                                                      dotdot, handle);
1476                         if (rc != 0) {
1477                                 int rc2;
1478
1479                                 rc2 = __mdd_index_delete(ctxt,
1480                                                          child, dot, handle);
1481                                 if (rc2 != 0)
1482                                         CERROR("Failure to cleanup after dotdot"
1483                                                " creation: %d (%d)\n", rc2, rc);
1484                                 else
1485                                         __mdd_ref_del(ctxt, child, handle);
1486                         }
1487                 }
1488         }
1489         RETURN(rc);
1490 }
1491
1492 static int mdd_create_data(const struct lu_context *ctxt,
1493                            struct md_object *pobj, struct md_object *cobj,
1494                            const struct md_create_spec *spec,
1495                            struct md_attr *ma)
1496 {
1497         struct mdd_device *mdd = mdo2mdd(pobj);
1498         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1499         struct mdd_object *son = md2mdd_obj(cobj);
1500         struct lu_attr    *attr = &ma->ma_attr;
1501         struct lov_mds_md *lmm = NULL;
1502         int                lmm_size = 0;
1503         struct thandle    *handle;
1504         int                rc;
1505         ENTRY;
1506
1507         rc = mdd_lov_create(ctxt, mdd, mdd_pobj, son, &lmm, &lmm_size, spec,
1508                             attr);
1509         if (rc)
1510                 RETURN(rc);
1511
1512         mdd_txn_param_build(ctxt, &MDD_TXN_CREATE_DATA);
1513         handle = mdd_trans_start(ctxt, mdd);
1514         if (IS_ERR(handle))
1515                 RETURN(PTR_ERR(handle));
1516
1517         /*XXX: setting the lov ea is not locked but setting the attr is locked? */
1518         if (rc == 0) {
1519                 rc = mdd_lov_set_md(ctxt, mdd_pobj, son, lmm, lmm_size,
1520                                     handle, 0);
1521                 if (rc == 0)
1522                         rc = mdd_attr_get_internal_locked(ctxt, son, ma);
1523         }
1524
1525         mdd_trans_stop(ctxt, mdd, rc, handle);
1526         RETURN(rc);
1527 }
1528
1529 static int mdd_create_sanity_check(const struct lu_context *ctxt,
1530                                    struct mdd_device *mdd,
1531                                    struct md_object *pobj,
1532                                    const char *name, struct md_attr *ma)
1533 {
1534         struct mdd_thread_info *info = mdd_ctx_info(ctxt);
1535         struct lu_attr    *la        = &info->mti_la;
1536         struct lu_fid     *fid       = &info->mti_fid;
1537         struct mdd_object *obj       = md2mdd_obj(pobj);
1538         int rc;
1539
1540         ENTRY;
1541         /* EEXIST check */
1542         if (mdd_is_dead_obj(obj))
1543                 RETURN(-ENOENT);
1544         rc = mdd_lookup(ctxt, pobj, name, fid);
1545         if (rc != -ENOENT)
1546                 RETURN(rc ? : -EEXIST);
1547
1548         /* sgid check */
1549         mdd_read_lock(ctxt, obj);
1550         rc = __mdd_la_get(ctxt, obj, la);
1551         mdd_read_unlock(ctxt, obj);
1552         if (rc != 0)
1553                 RETURN(rc);
1554
1555         if (la->la_mode & S_ISGID) {
1556                 ma->ma_attr.la_gid = la->la_gid;
1557                 if (S_ISDIR(ma->ma_attr.la_mode)) {
1558                         ma->ma_attr.la_mode |= S_ISGID;
1559                         ma->ma_attr.la_valid |= LA_MODE;
1560                 }
1561         }
1562
1563         switch (ma->ma_attr.la_mode & S_IFMT) {
1564         case S_IFREG:
1565         case S_IFDIR:
1566         case S_IFLNK:
1567         case S_IFCHR:
1568         case S_IFBLK:
1569         case S_IFIFO:
1570         case S_IFSOCK:
1571                 rc = 0;
1572                 break;
1573         default:
1574                 rc = -EINVAL;
1575                 break;
1576         }
1577         RETURN(rc);
1578 }
1579
1580 /*
1581  * Create object and insert it into namespace.
1582  */
1583 static int mdd_create(const struct lu_context *ctxt, struct md_object *pobj,
1584                       const char *name, struct md_object *child,
1585                       const struct md_create_spec *spec,
1586                       struct md_attr* ma)
1587 {
1588         struct mdd_device *mdd = mdo2mdd(pobj);
1589         struct mdd_object *mdd_pobj = md2mdd_obj(pobj);
1590         struct mdd_object *son = md2mdd_obj(child);
1591         struct lu_attr *attr = &ma->ma_attr;
1592         struct lov_mds_md *lmm = NULL;
1593         struct thandle *handle;
1594         int rc, created = 0, inserted = 0, lmm_size = 0;
1595         ENTRY;
1596
1597         /* sanity checks before big job */
1598         rc = mdd_create_sanity_check(ctxt, mdd, pobj, name, ma);
1599         if (rc)
1600                 RETURN(rc);
1601         /* no RPC inside the transaction, so OST objects should be created at
1602          * first */
1603         if (S_ISREG(attr->la_mode)) {
1604                 rc = mdd_lov_create(ctxt, mdd, mdd_pobj, son, &lmm, &lmm_size,
1605                                     spec, attr);
1606                 if (rc)
1607                         RETURN(rc);
1608         }
1609
1610         mdd_txn_param_build(ctxt, &MDD_TXN_MKDIR);
1611         handle = mdd_trans_start(ctxt, mdd);
1612         if (IS_ERR(handle))
1613                 RETURN(PTR_ERR(handle));
1614
1615         mdd_write_lock(ctxt, mdd_pobj);
1616
1617         /*
1618          * XXX check that link can be added to the parent in mkdir case.
1619          */
1620
1621         /*
1622          * Two operations have to be performed:
1623          *
1624          *  - allocation of new object (->do_create()), and
1625          *
1626          *  - insertion into parent index (->dio_insert()).
1627          *
1628          * Due to locking, operation order is not important, when both are
1629          * successful, *but* error handling cases are quite different:
1630          *
1631          *  - if insertion is done first, and following object creation fails,
1632          *  insertion has to be rolled back, but this operation might fail
1633          *  also leaving us with dangling index entry.
1634          *
1635          *  - if creation is done first, is has to be undone if insertion
1636          *  fails, leaving us with leaked space, which is neither good, nor
1637          *  fatal.
1638          *
1639          * It seems that creation-first is simplest solution, but it is
1640          * sub-optimal in the frequent
1641          *
1642          *         $ mkdir foo
1643          *         $ mkdir foo
1644          *
1645          * case, because second mkdir is bound to create object, only to
1646          * destroy it immediately.
1647          *
1648          * Note that local file systems do
1649          *
1650          *     0. lookup -> -EEXIST
1651          *
1652          *     1. create
1653          *
1654          *     2. insert
1655          *
1656          * Maybe we should do the same. For now: creation-first.
1657          */
1658
1659         mdd_write_lock(ctxt, son);
1660         rc = __mdd_object_create(ctxt, son, ma, handle);
1661         if (rc) {
1662                 mdd_write_unlock(ctxt, son);
1663                 GOTO(cleanup, rc);
1664         }
1665
1666         created = 1;
1667
1668         rc = __mdd_object_initialize(ctxt, mdo2fid(mdd_pobj),
1669                                      son, ma, handle);
1670         mdd_write_unlock(ctxt, son);
1671         if (rc)
1672                 /*
1673                  * Object has no links, so it will be destroyed when last
1674                  * reference is released. (XXX not now.)
1675                  */
1676                 GOTO(cleanup, rc);
1677
1678         rc = __mdd_index_insert(ctxt, mdd_pobj, mdo2fid(son),
1679                                 name, S_ISDIR(attr->la_mode), handle);
1680
1681         if (rc)
1682                 GOTO(cleanup, rc);
1683
1684         inserted = 1;
1685         rc = mdd_lov_set_md(ctxt, mdd_pobj, son, lmm, lmm_size, handle, 0);
1686         if (rc) {
1687                 CERROR("error on stripe info copy %d \n", rc);
1688                 GOTO(cleanup, rc);
1689         }
1690
1691         if (S_ISLNK(attr->la_mode)) {
1692                 struct dt_object *dt = mdd_object_child(son);
1693                 const char *target_name = spec->u.sp_symname;
1694                 int sym_len = strlen(target_name);
1695                 loff_t pos = 0;
1696
1697                 rc = dt->do_body_ops->dbo_write(ctxt, dt, target_name,
1698                                                 sym_len, &pos, handle);
1699                 if (rc == sym_len)
1700                         rc = 0;
1701                 else
1702                         rc = -EFAULT;
1703         }
1704         /* return attr back */
1705         rc = mdd_attr_get_internal_locked(ctxt, son, ma);
1706 cleanup:
1707         if (rc && created) {
1708                 int rc2 = 0;
1709
1710                 if (inserted) {
1711                         rc2 = __mdd_index_delete(ctxt, mdd_pobj, name, handle);
1712                         if (rc2)
1713                                 CERROR("error can not cleanup destroy %d\n",
1714                                        rc2);
1715                 }
1716                 if (rc2 == 0)
1717                         __mdd_ref_del(ctxt, son, handle);
1718         }
1719         if (lmm)
1720                 OBD_FREE(lmm, lmm_size);
1721         mdd_write_unlock(ctxt, mdd_pobj);
1722         mdd_trans_stop(ctxt, mdd, rc, handle);
1723         RETURN(rc);
1724 }
1725 /* partial operation */
1726 static int mdd_object_create(const struct lu_context *ctxt,
1727                              struct md_object *obj,
1728                              const struct md_create_spec *spec,
1729                              struct md_attr *ma)
1730 {
1731
1732         struct mdd_device *mdd = mdo2mdd(obj);
1733         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1734         struct thandle *handle;
1735         int rc;
1736         ENTRY;
1737
1738         mdd_txn_param_build(ctxt, &MDD_TXN_OBJECT_CREATE);
1739         handle = mdd_trans_start(ctxt, mdd);
1740         if (IS_ERR(handle))
1741                 RETURN(PTR_ERR(handle));
1742
1743         mdd_write_lock(ctxt, mdd_obj);
1744         rc = __mdd_object_create(ctxt, mdd_obj, ma, handle);
1745         if (rc == 0)
1746                 rc = __mdd_object_initialize(ctxt, spec->u.sp_pfid, mdd_obj,
1747                                      ma, handle);
1748         mdd_write_unlock(ctxt, mdd_obj);
1749
1750         if (rc == 0)
1751                 rc = mdd_attr_get_internal_locked(ctxt, mdd_obj, ma);
1752
1753         mdd_trans_stop(ctxt, mdd, rc, handle);
1754         RETURN(rc);
1755 }
1756 /* partial operation */
1757 static int mdd_name_insert(const struct lu_context *ctxt,
1758                            struct md_object *pobj, const char *name,
1759                            const struct lu_fid *fid, int isdir)
1760 {
1761         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1762         struct thandle *handle;
1763         int rc;
1764         ENTRY;
1765
1766         mdd_txn_param_build(ctxt, &MDD_TXN_INDEX_INSERT);
1767         handle = mdd_trans_start(ctxt, mdo2mdd(pobj));
1768         if (IS_ERR(handle))
1769                 RETURN(PTR_ERR(handle));
1770
1771         mdd_write_lock(ctxt, mdd_obj);
1772         rc = __mdd_index_insert(ctxt, mdd_obj, fid, name, isdir, handle);
1773         mdd_write_unlock(ctxt, mdd_obj);
1774
1775         mdd_trans_stop(ctxt, mdo2mdd(pobj), rc, handle);
1776         RETURN(rc);
1777 }
1778
1779 static int mdd_name_remove(const struct lu_context *ctxt,
1780                            struct md_object *pobj,
1781                            const char *name)
1782 {
1783         struct mdd_device *mdd = mdo2mdd(pobj);
1784         struct mdd_object *mdd_obj = md2mdd_obj(pobj);
1785         struct thandle *handle;
1786         int rc;
1787         ENTRY;
1788
1789         mdd_txn_param_build(ctxt, &MDD_TXN_INDEX_DELETE);
1790         handle = mdd_trans_start(ctxt, mdd);
1791         if (IS_ERR(handle))
1792                 RETURN(PTR_ERR(handle));
1793
1794         mdd_write_lock(ctxt, mdd_obj);
1795
1796         rc = __mdd_index_delete(ctxt, mdd_obj, name, handle);
1797
1798         mdd_write_unlock(ctxt, mdd_obj);
1799
1800         mdd_trans_stop(ctxt, mdd, rc, handle);
1801         RETURN(rc);
1802 }
1803
1804 static int mdd_rename_tgt(const struct lu_context *ctxt, struct md_object *pobj,
1805                           struct md_object *tobj, const struct lu_fid *lf,
1806                           const char *name)
1807 {
1808         struct mdd_device *mdd = mdo2mdd(pobj);
1809         struct mdd_object *mdd_tpobj = md2mdd_obj(pobj);
1810         struct mdd_object *mdd_tobj = NULL;
1811         struct thandle *handle;
1812         int rc;
1813         ENTRY;
1814
1815         mdd_txn_param_build(ctxt, &MDD_TXN_RENAME);
1816         handle = mdd_trans_start(ctxt, mdd);
1817         if (IS_ERR(handle))
1818                 RETURN(PTR_ERR(handle));
1819
1820         if (tobj)
1821                 mdd_tobj = md2mdd_obj(tobj);
1822
1823         mdd_lock2(ctxt, mdd_tpobj, mdd_tobj);
1824
1825         /*TODO rename sanity checking*/
1826         if (tobj) {
1827                 rc = __mdd_index_delete(ctxt, mdd_tpobj, name, handle);
1828                 if (rc)
1829                         GOTO(cleanup, rc);
1830         }
1831
1832         rc = __mdd_index_insert_only(ctxt, mdd_tpobj, lf, name, handle);
1833         if (rc)
1834                 GOTO(cleanup, rc);
1835
1836         if (tobj && lu_object_exists(&tobj->mo_lu))
1837                 __mdd_ref_del(ctxt, mdd_tobj, handle);
1838 cleanup:
1839         mdd_unlock2(ctxt, mdd_tpobj, mdd_tobj);
1840         mdd_trans_stop(ctxt, mdd, rc, handle);
1841         RETURN(rc);
1842 }
1843
1844 static int mdd_root_get(const struct lu_context *ctx,
1845                         struct md_device *m, struct lu_fid *f)
1846 {
1847         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1848
1849         ENTRY;
1850         *f = mdd->mdd_root_fid;
1851         RETURN(0);
1852 }
1853
1854 static int mdd_statfs(const struct lu_context *ctx,
1855                       struct md_device *m, struct kstatfs *sfs)
1856 {
1857         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1858         int rc;
1859
1860         ENTRY;
1861
1862         rc = mdd_child_ops(mdd)->dt_statfs(ctx, mdd->mdd_child, sfs);
1863
1864         RETURN(rc);
1865 }
1866
1867 static int mdd_get_maxsize(const struct lu_context *ctx,
1868                            struct md_device *m, int *md_size,
1869                            int *cookie_size)
1870 {
1871         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1872         ENTRY;
1873
1874         *md_size =  mdd_lov_mdsize(ctx, mdd);
1875         *cookie_size = mdd_lov_cookiesize(ctx, mdd);
1876
1877         RETURN(0);
1878 }
1879
1880 static void __mdd_ref_add(const struct lu_context *ctxt, struct mdd_object *obj,
1881                          struct thandle *handle)
1882 {
1883         struct dt_object *next;
1884
1885         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
1886         next = mdd_object_child(obj);
1887         next->do_ops->do_ref_add(ctxt, next, handle);
1888 }
1889
1890 static int mdd_ref_add(const struct lu_context *ctxt, struct md_object *obj)
1891 {
1892         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1893         struct mdd_device *mdd = mdo2mdd(obj);
1894         struct thandle *handle;
1895         ENTRY;
1896
1897         mdd_txn_param_build(ctxt, &MDD_TXN_XATTR_SET);
1898         handle = mdd_trans_start(ctxt, mdd);
1899         if (IS_ERR(handle))
1900                 RETURN(-ENOMEM);
1901
1902         mdd_write_lock(ctxt, mdd_obj);
1903         __mdd_ref_add(ctxt, mdd_obj, handle);
1904         mdd_write_unlock(ctxt, mdd_obj);
1905
1906         mdd_trans_stop(ctxt, mdd, 0, handle);
1907
1908         RETURN(0);
1909 }
1910
1911 static void
1912 __mdd_ref_del(const struct lu_context *ctxt, struct mdd_object *obj,
1913               struct thandle *handle)
1914 {
1915         struct dt_object *next = mdd_object_child(obj);
1916
1917         LASSERT(lu_object_exists(mdd2lu_obj(obj)));
1918
1919         next->do_ops->do_ref_del(ctxt, next, handle);
1920 }
1921
1922 static int mdd_ref_del(const struct lu_context *ctxt, struct md_object *obj,
1923                        struct md_attr *ma)
1924 {
1925         struct mdd_object *mdd_obj = md2mdd_obj(obj);
1926         struct mdd_device *mdd = mdo2mdd(obj);
1927         struct thandle *handle;
1928         int isdir;
1929         int rc;
1930         ENTRY;
1931
1932         mdd_txn_param_build(ctxt, &MDD_TXN_XATTR_SET);
1933         handle = mdd_trans_start(ctxt, mdd);
1934         if (IS_ERR(handle))
1935                 RETURN(-ENOMEM);
1936
1937         mdd_write_lock(ctxt, mdd_obj);
1938
1939         isdir = S_ISDIR(lu_object_attr(&obj->mo_lu));
1940         /* rmdir checks */
1941         if (isdir && dt_try_as_dir(ctxt, mdd_object_child(mdd_obj))) {
1942                 rc = mdd_dir_is_empty(ctxt, mdd_obj);
1943                 if (rc != 0)
1944                         GOTO(cleanup, rc);
1945         }
1946
1947         __mdd_ref_del(ctxt, mdd_obj, handle);
1948
1949         if (isdir) {
1950                 /* unlink dot */
1951                 __mdd_ref_del(ctxt, mdd_obj, handle);
1952         }
1953
1954         rc = __mdd_finish_unlink(ctxt, mdd_obj, ma);
1955
1956 cleanup:
1957         mdd_write_unlock(ctxt, mdd_obj);
1958         mdd_trans_stop(ctxt, mdd, rc, handle);
1959         RETURN(rc);
1960 }
1961
1962 /* do NOT or the MAY_*'s, you'll get the weakest */
1963 static int accmode(struct mdd_object *mdd_obj, int flags)
1964 {
1965         int res = 0;
1966
1967 #if 0
1968         /* Sadly, NFSD reopens a file repeatedly during operation, so the
1969          * "acc_mode = 0" allowance for newly-created files isn't honoured.
1970          * NFSD uses the MDS_OPEN_OWNEROVERRIDE flag to say that a file
1971          * owner can write to a file even if it is marked readonly to hide
1972          * its brokenness. (bug 5781) */
1973         if (flags & MDS_OPEN_OWNEROVERRIDE && inode->i_uid == current->fsuid)
1974                 return 0;
1975 #endif
1976         if (flags & FMODE_READ)
1977                 res = MAY_READ;
1978         if (flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
1979                 res |= MAY_WRITE;
1980         if (flags & MDS_FMODE_EXEC)
1981                 res = MAY_EXEC;
1982         return res;
1983 }
1984
1985 static int mdd_open(const struct lu_context *ctxt, struct md_object *obj,
1986                     int flags)
1987 {
1988         int mode = accmode(md2mdd_obj(obj), flags);
1989
1990         if (mode & MAY_WRITE) {
1991                 if (mdd_is_immutable(md2mdd_obj(obj)))
1992                         RETURN(-EACCES);
1993         }
1994
1995         atomic_inc(&md2mdd_obj(obj)->mod_count);
1996         return 0;
1997 }
1998
1999 static int mdd_close(const struct lu_context *ctxt, struct md_object *obj,
2000                      struct md_attr *ma)
2001 {
2002         int rc;
2003         struct mdd_object *mdd_obj;
2004
2005         mdd_obj = md2mdd_obj(obj);
2006         if (atomic_dec_and_test(&mdd_obj->mod_count)) {
2007                 /*TODO: Remove it from orphan list */
2008         }
2009
2010         mdd_read_lock(ctxt, mdd_obj);
2011         rc = __mdd_finish_unlink(ctxt, mdd_obj, ma);
2012         mdd_read_unlock(ctxt, mdd_obj);
2013         return rc;
2014 }
2015
2016 static int mdd_readpage(const struct lu_context *ctxt, struct md_object *obj,
2017                         const struct lu_rdpg *rdpg)
2018 {
2019         struct dt_object *next;
2020         struct mdd_object *mdd_obj = md2mdd_obj(obj);
2021         int rc;
2022
2023         LASSERT(lu_object_exists(mdd2lu_obj(mdd_obj)));
2024         next = mdd_object_child(mdd_obj);
2025
2026         mdd_read_lock(ctxt, mdd_obj);
2027         if (S_ISDIR(mdd_object_type(mdd_obj)) &&
2028             dt_try_as_dir(ctxt, next))
2029                 rc = next->do_ops->do_readpage(ctxt, next, rdpg);
2030         else
2031                 rc = -ENOTDIR;
2032         mdd_read_unlock(ctxt, mdd_obj);
2033         return rc;
2034 }
2035
2036 struct md_device_operations mdd_ops = {
2037         .mdo_root_get       = mdd_root_get,
2038         .mdo_statfs         = mdd_statfs,
2039         .mdo_get_maxsize    = mdd_get_maxsize,
2040 };
2041
2042 static struct md_dir_operations mdd_dir_ops = {
2043         .mdo_lookup        = mdd_lookup,
2044         .mdo_create        = mdd_create,
2045         .mdo_rename        = mdd_rename,
2046         .mdo_link          = mdd_link,
2047         .mdo_unlink        = mdd_unlink,
2048         .mdo_name_insert   = mdd_name_insert,
2049         .mdo_name_remove   = mdd_name_remove,
2050         .mdo_rename_tgt    = mdd_rename_tgt,
2051         .mdo_create_data   = mdd_create_data
2052 };
2053
2054
2055 static struct md_object_operations mdd_obj_ops = {
2056         .moo_attr_get      = mdd_attr_get,
2057         .moo_attr_set      = mdd_attr_set,
2058         .moo_xattr_get     = mdd_xattr_get,
2059         .moo_xattr_set     = mdd_xattr_set,
2060         .moo_xattr_list    = mdd_xattr_list,
2061         .moo_xattr_del     = mdd_xattr_del,
2062         .moo_object_create = mdd_object_create,
2063         .moo_ref_add       = mdd_ref_add,
2064         .moo_ref_del       = mdd_ref_del,
2065         .moo_open          = mdd_open,
2066         .moo_close         = mdd_close,
2067         .moo_readpage      = mdd_readpage,
2068         .moo_readlink      = mdd_readlink
2069 };
2070
2071 static struct obd_ops mdd_obd_device_ops = {
2072         .o_owner = THIS_MODULE
2073 };
2074
2075 static struct lu_device *mdd_device_alloc(const struct lu_context *ctx,
2076                                           struct lu_device_type *t,
2077                                           struct lustre_cfg *lcfg)
2078 {
2079         struct lu_device  *l;
2080         struct mdd_device *m;
2081
2082         OBD_ALLOC_PTR(m);
2083         if (m == NULL) {
2084                 l = ERR_PTR(-ENOMEM);
2085         } else {
2086                 md_device_init(&m->mdd_md_dev, t);
2087                 l = mdd2lu_dev(m);
2088                 l->ld_ops = &mdd_lu_ops;
2089                 m->mdd_md_dev.md_ops = &mdd_ops;
2090         }
2091
2092         return l;
2093 }
2094
2095 static void mdd_device_free(const struct lu_context *ctx,
2096                             struct lu_device *lu)
2097 {
2098         struct mdd_device *m = lu2mdd_dev(lu);
2099
2100         LASSERT(atomic_read(&lu->ld_ref) == 0);
2101         md_device_fini(&m->mdd_md_dev);
2102         OBD_FREE_PTR(m);
2103 }
2104
2105 static int mdd_type_init(struct lu_device_type *t)
2106 {
2107         return lu_context_key_register(&mdd_thread_key);
2108 }
2109
2110 static void mdd_type_fini(struct lu_device_type *t)
2111 {
2112         lu_context_key_degister(&mdd_thread_key);
2113 }
2114
2115 static struct lu_device_type_operations mdd_device_type_ops = {
2116         .ldto_init = mdd_type_init,
2117         .ldto_fini = mdd_type_fini,
2118
2119         .ldto_device_alloc = mdd_device_alloc,
2120         .ldto_device_free  = mdd_device_free,
2121
2122         .ldto_device_init    = mdd_device_init,
2123         .ldto_device_fini    = mdd_device_fini
2124 };
2125
2126 static struct lu_device_type mdd_device_type = {
2127         .ldt_tags     = LU_DEVICE_MD,
2128         .ldt_name     = LUSTRE_MDD0_NAME,
2129         .ldt_ops      = &mdd_device_type_ops,
2130         .ldt_ctx_tags = LCT_MD_THREAD
2131 };
2132
2133 static void *mdd_key_init(const struct lu_context *ctx,
2134                           struct lu_context_key *key)
2135 {
2136         struct mdd_thread_info *info;
2137
2138         OBD_ALLOC_PTR(info);
2139         if (info == NULL)
2140                 info = ERR_PTR(-ENOMEM);
2141         return info;
2142 }
2143
2144 static void mdd_key_fini(const struct lu_context *ctx,
2145                          struct lu_context_key *key, void *data)
2146 {
2147         struct mdd_thread_info *info = data;
2148         OBD_FREE_PTR(info);
2149 }
2150
2151 static struct lu_context_key mdd_thread_key = {
2152         .lct_tags = LCT_MD_THREAD,
2153         .lct_init = mdd_key_init,
2154         .lct_fini = mdd_key_fini
2155 };
2156
2157 struct lprocfs_vars lprocfs_mdd_obd_vars[] = {
2158         { 0 }
2159 };
2160
2161 struct lprocfs_vars lprocfs_mdd_module_vars[] = {
2162         { 0 }
2163 };
2164
2165 LPROCFS_INIT_VARS(mdd, lprocfs_mdd_module_vars, lprocfs_mdd_obd_vars);
2166
2167 static int __init mdd_mod_init(void)
2168 {
2169         struct lprocfs_static_vars lvars;
2170
2171         lprocfs_init_vars(mdd, &lvars);
2172         return class_register_type(&mdd_obd_device_ops, NULL, lvars.module_vars,
2173                                    LUSTRE_MDD0_NAME, &mdd_device_type);
2174 }
2175
2176 static void __exit mdd_mod_exit(void)
2177 {
2178         class_unregister_type(LUSTRE_MDD0_NAME);
2179 }
2180
2181 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2182 MODULE_DESCRIPTION("Lustre Meta-data Device Prototype ("LUSTRE_MDD0_NAME")");
2183 MODULE_LICENSE("GPL");
2184
2185 cfs_module(mdd, "0.1.0", mdd_mod_init, mdd_mod_exit);