lustre/osd-ldiskfs/osd_handler.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  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/osd/osd_handler.c
  37  *
  38  * Top-level entry points into osd module
  39  *
  40  * Author: Nikita Danilov <nikita@clusterfs.com>
  41  *         Pravin Shelar <pravin.shelar@sun.com> : Added fid in dirent
  42  */
  43
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47 #define DEBUG_SUBSYSTEM S_MDS
  48
  49 #include <linux/module.h>
  50
  51 /* LUSTRE_VERSION_CODE */
  52 #include <lustre_ver.h>
  53 /* prerequisite for linux/xattr.h */
  54 #include <linux/types.h>
  55 /* prerequisite for linux/xattr.h */
  56 #include <linux/fs.h>
  57 /* XATTR_{REPLACE,CREATE} */
  58 #include <linux/xattr.h>
  59 /* simple_mkdir() */
  60 #include <lvfs.h>
  61
  62 /*
  63  * struct OBD_{ALLOC,FREE}*()
  64  * OBD_FAIL_CHECK
  65  */
  66 #include <obd_support.h>
  67 /* struct ptlrpc_thread */
  68 #include <lustre_net.h>
  69
  70 /* fid_is_local() */
  71 #include <lustre_fid.h>
  72
  73 #include "osd_internal.h"
  74 #include "osd_igif.h"
  75
  76 /* llo_* api support */
  77 #include <md_object.h>
  78
  79 static const char dot[] = ".";
  80 static const char dotdot[] = "..";
  81 static const char remote_obj_dir[] = "REM_OBJ_DIR";
  82
  83 struct osd_directory {
  84         struct iam_container od_container;
  85         struct iam_descr     od_descr;
  86 };
  87
  88 struct osd_object {
  89         struct dt_object       oo_dt;
  90         /**
  91          * Inode for file system object represented by this osd_object. This
  92          * inode is pinned for the whole duration of lu_object life.
  93          *
  94          * Not modified concurrently (either setup early during object
  95          * creation, or assigned by osd_object_create() under write lock).
  96          */
  97         struct inode          *oo_inode;
  98         /**
  99          * to protect index ops.
 100          */
 101         cfs_rw_semaphore_t     oo_ext_idx_sem;
 102         cfs_rw_semaphore_t     oo_sem;
 103         struct osd_directory  *oo_dir;
 104         /** protects inode attributes. */
 105         cfs_spinlock_t         oo_guard;
 106         /**
 107          * Following two members are used to indicate the presence of dot and
 108          * dotdot in the given directory. This is required for interop mode
 109          * (b11826).
 110          */
 111         int                    oo_compat_dot_created;
 112         int                    oo_compat_dotdot_created;
 113
 114         const struct lu_env   *oo_owner;
 115 #ifdef CONFIG_LOCKDEP
 116         struct lockdep_map     oo_dep_map;
 117 #endif
 118 };
 119
 120 static const struct lu_object_operations      osd_lu_obj_ops;
 121 static const struct lu_device_operations      osd_lu_ops;
 122 static       struct lu_context_key            osd_key;
 123 static const struct dt_object_operations      osd_obj_ops;
 124 static const struct dt_object_operations      osd_obj_ea_ops;
 125 static const struct dt_body_operations        osd_body_ops;
 126 static const struct dt_index_operations       osd_index_iam_ops;
 127 static const struct dt_index_operations       osd_index_ea_ops;
 128
 129 struct osd_thandle {
 130         struct thandle          ot_super;
 131         handle_t               *ot_handle;
 132         struct journal_callback ot_jcb;
 133         /* Link to the device, for debugging. */
 134         struct lu_ref_link     *ot_dev_link;
 135
 136 #if OSD_THANDLE_STATS
 137         /** time when this handle was allocated */
 138         cfs_time_t oth_alloced;
 139
 140         /** time when this thanle was started */
 141         cfs_time_t oth_started;
 142 #endif
 143 };
 144
 145 /*
 146  * Helpers.
 147  */
 148 static int lu_device_is_osd(const struct lu_device *d)
 149 {
 150         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &osd_lu_ops);
 151 }
 152
 153 static struct osd_device *osd_dt_dev(const struct dt_device *d)
 154 {
 155         LASSERT(lu_device_is_osd(&d->dd_lu_dev));
 156         return container_of0(d, struct osd_device, od_dt_dev);
 157 }
 158
 159 static struct osd_device *osd_dev(const struct lu_device *d)
 160 {
 161         LASSERT(lu_device_is_osd(d));
 162         return osd_dt_dev(container_of0(d, struct dt_device, dd_lu_dev));
 163 }
 164
 165 static struct osd_device *osd_obj2dev(const struct osd_object *o)
 166 {
 167         return osd_dev(o->oo_dt.do_lu.lo_dev);
 168 }
 169
 170 static struct super_block *osd_sb(const struct osd_device *dev)
 171 {
 172         return dev->od_mount->lmi_mnt->mnt_sb;
 173 }
 174
 175 static int osd_object_is_root(const struct osd_object *obj)
 176 {
 177         return osd_sb(osd_obj2dev(obj))->s_root->d_inode == obj->oo_inode;
 178 }
 179
 180 static struct osd_object *osd_obj(const struct lu_object *o)
 181 {
 182         LASSERT(lu_device_is_osd(o->lo_dev));
 183         return container_of0(o, struct osd_object, oo_dt.do_lu);
 184 }
 185
 186 static struct osd_object *osd_dt_obj(const struct dt_object *d)
 187 {
 188         return osd_obj(&d->do_lu);
 189 }
 190
 191 static struct lu_device *osd2lu_dev(struct osd_device *osd)
 192 {
 193         return &osd->od_dt_dev.dd_lu_dev;
 194 }
 195
 196 static journal_t *osd_journal(const struct osd_device *dev)
 197 {
 198         return LDISKFS_SB(osd_sb(dev))->s_journal;
 199 }
 200
 201 static int osd_has_index(const struct osd_object *obj)
 202 {
 203         return obj->oo_dt.do_index_ops != NULL;
 204 }
 205
 206 static int osd_object_invariant(const struct lu_object *l)
 207 {
 208         return osd_invariant(osd_obj(l));
 209 }
 210
 211 #ifdef HAVE_QUOTA_SUPPORT
 212 static inline void
 213 osd_push_ctxt(const struct lu_env *env, struct osd_ctxt *save)
 214 {
 215         struct md_ucred    *uc = md_ucred(env);
 216
 217         LASSERT(uc != NULL);
 218
 219         save->oc_uid = current->fsuid;
 220         save->oc_gid = current->fsgid;
 221         save->oc_cap = current->cap_effective;
 222         current->fsuid         = uc->mu_fsuid;
 223         current->fsgid         = uc->mu_fsgid;
 224         current->cap_effective = uc->mu_cap;
 225 }
 226
 227 static inline void
 228 osd_pop_ctxt(struct osd_ctxt *save)
 229 {
 230         current->fsuid         = save->oc_uid;
 231         current->fsgid         = save->oc_gid;
 232         current->cap_effective = save->oc_cap;
 233 }
 234 #endif
 235
 236 static inline struct osd_thread_info *osd_oti_get(const struct lu_env *env)
 237 {
 238         return lu_context_key_get(&env->le_ctx, &osd_key);
 239 }
 240
 241 /*
 242  * Concurrency: doesn't matter
 243  */
 244 static int osd_read_locked(const struct lu_env *env, struct osd_object *o)
 245 {
 246         return osd_oti_get(env)->oti_r_locks > 0;
 247 }
 248
 249 /*
 250  * Concurrency: doesn't matter
 251  */
 252 static int osd_write_locked(const struct lu_env *env, struct osd_object *o)
 253 {
 254         struct osd_thread_info *oti = osd_oti_get(env);
 255         return oti->oti_w_locks > 0 && o->oo_owner == env;
 256 }
 257
 258 /*
 259  * Concurrency: doesn't access mutable data
 260  */
 261 static int osd_root_get(const struct lu_env *env,
 262                         struct dt_device *dev, struct lu_fid *f)
 263 {
 264         struct inode *inode;
 265
 266         inode = osd_sb(osd_dt_dev(dev))->s_root->d_inode;
 267         LU_IGIF_BUILD(f, inode->i_ino, inode->i_generation);
 268         return 0;
 269 }
 270
 271 /*
 272  * OSD object methods.
 273  */
 274
 275 /*
 276  * Concurrency: no concurrent access is possible that early in object
 277  * life-cycle.
 278  */
 279 static struct lu_object *osd_object_alloc(const struct lu_env *env,
 280                                           const struct lu_object_header *hdr,
 281                                           struct lu_device *d)
 282 {
 283         struct osd_object *mo;
 284
 285         OBD_ALLOC_PTR(mo);
 286         if (mo != NULL) {
 287                 struct lu_object *l;
 288
 289                 l = &mo->oo_dt.do_lu;
 290                 dt_object_init(&mo->oo_dt, NULL, d);
 291                 if (osd_dev(d)->od_iop_mode)
 292                         mo->oo_dt.do_ops = &osd_obj_ea_ops;
 293                 else
 294                         mo->oo_dt.do_ops = &osd_obj_ops;
 295
 296                 l->lo_ops = &osd_lu_obj_ops;
 297                 cfs_init_rwsem(&mo->oo_sem);
 298                 cfs_init_rwsem(&mo->oo_ext_idx_sem);
 299                 cfs_spin_lock_init(&mo->oo_guard);
 300                 return l;
 301         } else
 302                 return NULL;
 303 }
 304
 305 /*
 306  * retrieve object from backend ext fs.
 307  **/
 308 static struct inode *osd_iget(struct osd_thread_info *info,
 309                               struct osd_device *dev,
 310                               const struct osd_inode_id *id)
 311 {
 312         struct inode *inode = NULL;
 313
 314 #ifdef HAVE_EXT4_LDISKFS
 315         inode = ldiskfs_iget(osd_sb(dev), id->oii_ino);
 316         if (IS_ERR(inode))
 317         /* Newer kernels return an error instead of a NULL pointer */
 318                 inode = NULL;
 319 #else
 320         inode = iget(osd_sb(dev), id->oii_ino);
 321 #endif
 322         if (inode == NULL) {
 323                 CERROR("no inode\n");
 324                 inode = ERR_PTR(-EACCES);
 325         } else if (id->oii_gen != OSD_OII_NOGEN &&
 326                    inode->i_generation != id->oii_gen) {
 327                 iput(inode);
 328                 inode = ERR_PTR(-ESTALE);
 329         } else if (inode->i_nlink == 0) {
 330                 /* due to parallel readdir and unlink,
 331                 * we can have dead inode here. */
 332                 CWARN("stale inode\n");
 333                 make_bad_inode(inode);
 334                 iput(inode);
 335                 inode = ERR_PTR(-ESTALE);
 336         } else if (is_bad_inode(inode)) {
 337                 CERROR("bad inode %lx\n",inode->i_ino);
 338                 iput(inode);
 339                 inode = ERR_PTR(-ENOENT);
 340         }
 341         return inode;
 342 }
 343
 344 static int osd_fid_lookup(const struct lu_env *env,
 345                           struct osd_object *obj, const struct lu_fid *fid)
 346 {
 347         struct osd_thread_info *info;
 348         struct lu_device       *ldev = obj->oo_dt.do_lu.lo_dev;
 349         struct osd_device      *dev;
 350         struct osd_inode_id    *id;
 351         struct osd_oi          *oi;
 352         struct inode           *inode;
 353         int                     result;
 354
 355         LINVRNT(osd_invariant(obj));
 356         LASSERT(obj->oo_inode == NULL);
 357         LASSERT(fid_is_sane(fid) || osd_fid_is_root(fid));
 358         /*
 359          * This assertion checks that osd layer sees only local
 360          * fids. Unfortunately it is somewhat expensive (does a
 361          * cache-lookup). Disabling it for production/acceptance-testing.
 362          */
 363         LASSERT(1 || fid_is_local(env, ldev->ld_site, fid));
 364
 365         ENTRY;
 366
 367         info = osd_oti_get(env);
 368         dev  = osd_dev(ldev);
 369         id   = &info->oti_id;
 370         oi   = &dev->od_oi;
 371
 372         if (OBD_FAIL_CHECK(OBD_FAIL_OST_ENOENT))
 373                 RETURN(-ENOENT);
 374
 375         result = osd_oi_lookup(info, oi, fid, id);
 376         if (result == 0) {
 377                 inode = osd_iget(info, dev, id);
 378                 if (!IS_ERR(inode)) {
 379                         obj->oo_inode = inode;
 380                         LASSERT(obj->oo_inode->i_sb == osd_sb(dev));
 381                         if (dev->od_iop_mode) {
 382                                 obj->oo_compat_dot_created = 1;
 383                                 obj->oo_compat_dotdot_created = 1;
 384                         }
 385                         result = 0;
 386                 } else
 387                         /*
 388                          * If fid wasn't found in oi, inode-less object is
 389                          * created, for which lu_object_exists() returns
 390                          * false. This is used in a (frequent) case when
 391                          * objects are created as locking anchors or
 392                          * place holders for objects yet to be created.
 393                          */
 394                         result = PTR_ERR(inode);
 395         } else if (result == -ENOENT)
 396                 result = 0;
 397         LINVRNT(osd_invariant(obj));
 398
 399         RETURN(result);
 400 }
 401
 402 /*
 403  * Concurrency: shouldn't matter.
 404  */
 405 static void osd_object_init0(struct osd_object *obj)
 406 {
 407         LASSERT(obj->oo_inode != NULL);
 408         obj->oo_dt.do_body_ops = &osd_body_ops;
 409         obj->oo_dt.do_lu.lo_header->loh_attr |=
 410                 (LOHA_EXISTS | (obj->oo_inode->i_mode & S_IFMT));
 411 }
 412
 413 /*
 414  * Concurrency: no concurrent access is possible that early in object
 415  * life-cycle.
 416  */
 417 static int osd_object_init(const struct lu_env *env, struct lu_object *l,
 418                            const struct lu_object_conf *unused)
 419 {
 420         struct osd_object *obj = osd_obj(l);
 421         int result;
 422
 423         LINVRNT(osd_invariant(obj));
 424
 425         result = osd_fid_lookup(env, obj, lu_object_fid(l));
 426         if (result == 0) {
 427                 if (obj->oo_inode != NULL)
 428                         osd_object_init0(obj);
 429         }
 430         LINVRNT(osd_invariant(obj));
 431         return result;
 432 }
 433
 434 /*
 435  * Concurrency: no concurrent access is possible that late in object
 436  * life-cycle.
 437  */
 438 static void osd_object_free(const struct lu_env *env, struct lu_object *l)
 439 {
 440         struct osd_object *obj = osd_obj(l);
 441
 442         LINVRNT(osd_invariant(obj));
 443
 444         dt_object_fini(&obj->oo_dt);
 445         OBD_FREE_PTR(obj);
 446 }
 447
 448 /**
 449  * IAM Iterator
 450  */
 451 static struct iam_path_descr *osd_it_ipd_get(const struct lu_env *env,
 452                                              const struct iam_container *bag)
 453 {
 454         return bag->ic_descr->id_ops->id_ipd_alloc(bag,
 455                                            osd_oti_get(env)->oti_it_ipd);
 456 }
 457
 458 static struct iam_path_descr *osd_idx_ipd_get(const struct lu_env *env,
 459                                               const struct iam_container *bag)
 460 {
 461         return bag->ic_descr->id_ops->id_ipd_alloc(bag,
 462                                            osd_oti_get(env)->oti_idx_ipd);
 463 }
 464
 465 static void osd_ipd_put(const struct lu_env *env,
 466                         const struct iam_container *bag,
 467                         struct iam_path_descr *ipd)
 468 {
 469         bag->ic_descr->id_ops->id_ipd_free(ipd);
 470 }
 471
 472 /*
 473  * Concurrency: no concurrent access is possible that late in object
 474  * life-cycle.
 475  */
 476 static void osd_index_fini(struct osd_object *o)
 477 {
 478         struct iam_container *bag;
 479
 480         if (o->oo_dir != NULL) {
 481                 bag = &o->oo_dir->od_container;
 482                 if (o->oo_inode != NULL) {
 483                         if (bag->ic_object == o->oo_inode)
 484                                 iam_container_fini(bag);
 485                 }
 486                 OBD_FREE_PTR(o->oo_dir);
 487                 o->oo_dir = NULL;
 488         }
 489 }
 490
 491 /*
 492  * Concurrency: no concurrent access is possible that late in object
 493  * life-cycle (for all existing callers, that is. New callers have to provide
 494  * their own locking.)
 495  */
 496 static int osd_inode_unlinked(const struct inode *inode)
 497 {
 498         return inode->i_nlink == 0;
 499 }
 500
 501 enum {
 502         OSD_TXN_OI_DELETE_CREDITS    = 20,
 503         OSD_TXN_INODE_DELETE_CREDITS = 20
 504 };
 505
 506 /*
 507  * Journal
 508  */
 509
 510 #if OSD_THANDLE_STATS
 511 /**
 512  * Set time when the handle is allocated
 513  */
 514 static void osd_th_alloced(struct osd_thandle *oth)
 515 {
 516         oth->oth_alloced = cfs_time_current();
 517 }
 518
 519 /**
 520  * Set time when the handle started
 521  */
 522 static void osd_th_started(struct osd_thandle *oth)
 523 {
 524         oth->oth_started = cfs_time_current();
 525 }
 526
 527 /**
 528  * Helper function to convert time interval to microseconds packed in
 529  * long int (default time units for the counter in "stats" initialized
 530  * by lu_time_init() )
 531  */
 532 static long interval_to_usec(cfs_time_t start, cfs_time_t end)
 533 {
 534         struct timeval val;
 535
 536         cfs_duration_usec(cfs_time_sub(end, start), &val);
 537         return val.tv_sec * 1000000 + val.tv_usec;
 538 }
 539
 540 /**
 541  * Check whether the we deal with this handle for too long.
 542  */
 543 static void __osd_th_check_slow(void *oth, struct osd_device *dev,
 544                                 cfs_time_t alloced, cfs_time_t started,
 545                                 cfs_time_t closed)
 546 {
 547         cfs_time_t now = cfs_time_current();
 548
 549         LASSERT(dev != NULL);
 550
 551         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_STARTING,
 552                             interval_to_usec(alloced, started));
 553         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_OPEN,
 554                             interval_to_usec(started, closed));
 555         lprocfs_counter_add(dev->od_stats, LPROC_OSD_THANDLE_CLOSING,
 556                             interval_to_usec(closed, now));
 557
 558         if (cfs_time_before(cfs_time_add(alloced, cfs_time_seconds(30)), now)) {
 559                 CWARN("transaction handle %p was open for too long: "
 560                       "now "CFS_TIME_T" ,"
 561                       "alloced "CFS_TIME_T" ,"
 562                       "started "CFS_TIME_T" ,"
 563                       "closed "CFS_TIME_T"\n",
 564                       oth, now, alloced, started, closed);
 565                 libcfs_debug_dumpstack(NULL);
 566         }
 567 }
 568
 569 #define OSD_CHECK_SLOW_TH(oth, dev, expr)                               \
 570 {                                                                       \
 571         cfs_time_t __closed = cfs_time_current();                       \
 572         cfs_time_t __alloced = oth->oth_alloced;                        \
 573         cfs_time_t __started = oth->oth_started;                        \
 574                                                                         \
 575         expr;                                                           \
 576         __osd_th_check_slow(oth, dev, __alloced, __started, __closed);  \
 577 }
 578
 579 #else /* OSD_THANDLE_STATS */
 580
 581 #define osd_th_alloced(h)                  do {} while(0)
 582 #define osd_th_started(h)                  do {} while(0)
 583 #define OSD_CHECK_SLOW_TH(oth, dev, expr)  expr
 584
 585 #endif /* OSD_THANDLE_STATS */
 586
 587 /*
 588  * Concurrency: doesn't access mutable data.
 589  */
 590 static int osd_param_is_sane(const struct osd_device *dev,
 591                              const struct txn_param *param)
 592 {
 593         return param->tp_credits <= osd_journal(dev)->j_max_transaction_buffers;
 594 }
 595
 596 /*
 597  * Concurrency: shouldn't matter.
 598  */
 599 static void osd_trans_commit_cb(struct journal_callback *jcb, int error)
 600 {
 601         struct osd_thandle *oh = container_of0(jcb, struct osd_thandle, ot_jcb);
 602         struct thandle     *th  = &oh->ot_super;
 603         struct dt_device   *dev = th->th_dev;
 604         struct lu_device   *lud = &dev->dd_lu_dev;
 605
 606         LASSERT(dev != NULL);
 607         LASSERT(oh->ot_handle == NULL);
 608
 609         if (error) {
 610                 CERROR("transaction @0x%p commit error: %d\n", th, error);
 611         } else {
 612                 struct lu_env *env = &osd_dt_dev(dev)->od_env_for_commit;
 613                 /*
 614                  * This od_env_for_commit is only for commit usage.  see
 615                  * "struct dt_device"
 616                  */
 617                 lu_context_enter(&env->le_ctx);
 618                 dt_txn_hook_commit(env, th);
 619                 lu_context_exit(&env->le_ctx);
 620         }
 621
 622         lu_ref_del_at(&lud->ld_reference, oh->ot_dev_link, "osd-tx", th);
 623         lu_device_put(lud);
 624         th->th_dev = NULL;
 625
 626         lu_context_exit(&th->th_ctx);
 627         lu_context_fini(&th->th_ctx);
 628         OBD_FREE_PTR(oh);
 629 }
 630
 631 /*
 632  * Concurrency: shouldn't matter.
 633  */
 634 static struct thandle *osd_trans_start(const struct lu_env *env,
 635                                        struct dt_device *d,
 636                                        struct txn_param *p)
 637 {
 638         struct osd_device  *dev = osd_dt_dev(d);
 639         handle_t           *jh;
 640         struct osd_thandle *oh;
 641         struct thandle     *th;
 642         int hook_res;
 643
 644         ENTRY;
 645
 646         hook_res = dt_txn_hook_start(env, d, p);
 647         if (hook_res != 0)
 648                 RETURN(ERR_PTR(hook_res));
 649
 650         if (osd_param_is_sane(dev, p)) {
 651                 OBD_ALLOC_GFP(oh, sizeof *oh, CFS_ALLOC_IO);
 652                 if (oh != NULL) {
 653                         struct osd_thread_info *oti = osd_oti_get(env);
 654
 655                         /*
 656                          * XXX temporary stuff. Some abstraction layer should
 657                          * be used.
 658                          */
 659                         oti->oti_dev = dev;
 660                         osd_th_alloced(oh);
 661                         jh = ldiskfs_journal_start_sb(osd_sb(dev), p->tp_credits);
 662                         osd_th_started(oh);
 663                         if (!IS_ERR(jh)) {
 664                                 oh->ot_handle = jh;
 665                                 th = &oh->ot_super;
 666                                 th->th_dev = d;
 667                                 th->th_result = 0;
 668                                 jh->h_sync = p->tp_sync;
 669                                 lu_device_get(&d->dd_lu_dev);
 670                                 oh->ot_dev_link = lu_ref_add
 671                                         (&d->dd_lu_dev.ld_reference,
 672                                          "osd-tx", th);
 673                                 /* add commit callback */
 674                                 lu_context_init(&th->th_ctx, LCT_TX_HANDLE);
 675                                 lu_context_enter(&th->th_ctx);
 676                                 osd_journal_callback_set(jh, osd_trans_commit_cb,
 677                                                          (struct journal_callback *)&oh->ot_jcb);
 678                                         LASSERT(oti->oti_txns == 0);
 679                                         LASSERT(oti->oti_r_locks == 0);
 680                                         LASSERT(oti->oti_w_locks == 0);
 681                                         oti->oti_txns++;
 682                         } else {
 683                                 OBD_FREE_PTR(oh);
 684                                 th = (void *)jh;
 685                         }
 686                 } else
 687                         th = ERR_PTR(-ENOMEM);
 688         } else {
 689                 CERROR("Invalid transaction parameters\n");
 690                 th = ERR_PTR(-EINVAL);
 691         }
 692
 693         RETURN(th);
 694 }
 695
 696 /*
 697  * Concurrency: shouldn't matter.
 698  */
 699 static void osd_trans_stop(const struct lu_env *env, struct thandle *th)
 700 {
 701         int result;
 702         struct osd_thandle *oh;
 703         struct osd_thread_info *oti = osd_oti_get(env);
 704
 705         ENTRY;
 706
 707         oh = container_of0(th, struct osd_thandle, ot_super);
 708         if (oh->ot_handle != NULL) {
 709                 handle_t *hdl = oh->ot_handle;
 710
 711                 LASSERT(oti->oti_txns == 1);
 712                 oti->oti_txns--;
 713                 LASSERT(oti->oti_r_locks == 0);
 714                 LASSERT(oti->oti_w_locks == 0);
 715                 result = dt_txn_hook_stop(env, th);
 716                 if (result != 0)
 717                         CERROR("Failure in transaction hook: %d\n", result);
 718                 oh->ot_handle = NULL;
 719                 OSD_CHECK_SLOW_TH(oh, oti->oti_dev,
 720                                   result = ldiskfs_journal_stop(hdl));
 721                 if (result != 0)
 722                         CERROR("Failure to stop transaction: %d\n", result);
 723         }
 724         EXIT;
 725 }
 726
 727 /*
 728  * Concurrency: no concurrent access is possible that late in object
 729  * life-cycle.
 730  */
 731 static int osd_inode_remove(const struct lu_env *env, struct osd_object *obj)
 732 {
 733         const struct lu_fid    *fid = lu_object_fid(&obj->oo_dt.do_lu);
 734         struct osd_device      *osd = osd_obj2dev(obj);
 735         struct osd_thread_info *oti = osd_oti_get(env);
 736         struct txn_param       *prm = &oti->oti_txn;
 737         struct lu_env          *env_del_obj = &oti->oti_obj_delete_tx_env;
 738         struct thandle         *th;
 739         int result;
 740
 741         lu_env_init(env_del_obj, LCT_DT_THREAD);
 742         txn_param_init(prm, OSD_TXN_OI_DELETE_CREDITS +
 743                             OSD_TXN_INODE_DELETE_CREDITS);
 744         th = osd_trans_start(env_del_obj, &osd->od_dt_dev, prm);
 745         if (!IS_ERR(th)) {
 746                 result = osd_oi_delete(osd_oti_get(env_del_obj),
 747                                        &osd->od_oi, fid, th);
 748                 osd_trans_stop(env_del_obj, th);
 749         } else
 750                 result = PTR_ERR(th);
 751
 752         lu_env_fini(env_del_obj);
 753         return result;
 754 }
 755
 756 /*
 757  * Called just before object is freed. Releases all resources except for
 758  * object itself (that is released by osd_object_free()).
 759  *
 760  * Concurrency: no concurrent access is possible that late in object
 761  * life-cycle.
 762  */
 763 static void osd_object_delete(const struct lu_env *env, struct lu_object *l)
 764 {
 765         struct osd_object *obj   = osd_obj(l);
 766         struct inode      *inode = obj->oo_inode;
 767
 768         LINVRNT(osd_invariant(obj));
 769
 770         /*
 771          * If object is unlinked remove fid->ino mapping from object index.
 772          */
 773
 774         osd_index_fini(obj);
 775         if (inode != NULL) {
 776                 int result;
 777
 778                 if (osd_inode_unlinked(inode)) {
 779                         result = osd_inode_remove(env, obj);
 780                         if (result != 0)
 781                                 LU_OBJECT_DEBUG(D_ERROR, env, l,
 782                                                 "Failed to cleanup: %d\n",
 783                                                 result);
 784                 }
 785
 786                 iput(inode);
 787                 obj->oo_inode = NULL;
 788         }
 789 }
 790
 791 /*
 792  * Concurrency: ->loo_object_release() is called under site spin-lock.
 793  */
 794 static void osd_object_release(const struct lu_env *env,
 795                                struct lu_object *l)
 796 {
 797         struct osd_object *o = osd_obj(l);
 798
 799         LASSERT(!lu_object_is_dying(l->lo_header));
 800         if (o->oo_inode != NULL && osd_inode_unlinked(o->oo_inode))
 801                 cfs_set_bit(LU_OBJECT_HEARD_BANSHEE, &l->lo_header->loh_flags);
 802 }
 803
 804 /*
 805  * Concurrency: shouldn't matter.
 806  */
 807 static int osd_object_print(const struct lu_env *env, void *cookie,
 808                             lu_printer_t p, const struct lu_object *l)
 809 {
 810         struct osd_object *o = osd_obj(l);
 811         struct iam_descr  *d;
 812
 813         if (o->oo_dir != NULL)
 814                 d = o->oo_dir->od_container.ic_descr;
 815         else
 816                 d = NULL;
 817         return (*p)(env, cookie, LUSTRE_OSD_NAME"-object@%p(i:%p:%lu/%u)[%s]",
 818                     o, o->oo_inode,
 819                     o->oo_inode ? o->oo_inode->i_ino : 0UL,
 820                     o->oo_inode ? o->oo_inode->i_generation : 0,
 821                     d ? d->id_ops->id_name : "plain");
 822 }
 823
 824 /*
 825  * Concurrency: shouldn't matter.
 826  */
 827 int osd_statfs(const struct lu_env *env, struct dt_device *d,
 828                cfs_kstatfs_t *sfs)
 829 {
 830         struct osd_device *osd = osd_dt_dev(d);
 831         struct super_block *sb = osd_sb(osd);
 832         int result = 0;
 833
 834         cfs_spin_lock(&osd->od_osfs_lock);
 835         /* cache 1 second */
 836         if (cfs_time_before_64(osd->od_osfs_age, cfs_time_shift_64(-1))) {
 837                 result = ll_do_statfs(sb, &osd->od_kstatfs);
 838                 if (likely(result == 0)) /* N.B. statfs can't really fail */
 839                         osd->od_osfs_age = cfs_time_current_64();
 840         }
 841
 842         if (likely(result == 0))
 843                 *sfs = osd->od_kstatfs;
 844         cfs_spin_unlock(&osd->od_osfs_lock);
 845
 846         return result;
 847 }
 848
 849 /*
 850  * Concurrency: doesn't access mutable data.
 851  */
 852 static void osd_conf_get(const struct lu_env *env,
 853                          const struct dt_device *dev,
 854                          struct dt_device_param *param)
 855 {
 856         /*
 857          * XXX should be taken from not-yet-existing fs abstraction layer.
 858          */
 859         param->ddp_max_name_len  = LDISKFS_NAME_LEN;
 860         param->ddp_max_nlink     = LDISKFS_LINK_MAX;
 861         param->ddp_block_shift   = osd_sb(osd_dt_dev(dev))->s_blocksize_bits;
 862 }
 863
 864 /**
 865  * Helper function to get and fill the buffer with input values.
 866  */
 867 static struct lu_buf *osd_buf_get(const struct lu_env *env, void *area, ssize_t len)
 868 {
 869         struct lu_buf *buf;
 870
 871         buf = &osd_oti_get(env)->oti_buf;
 872         buf->lb_buf = area;
 873         buf->lb_len = len;
 874         return buf;
 875 }
 876
 877 /*
 878  * Concurrency: shouldn't matter.
 879  */
 880 static int osd_sync(const struct lu_env *env, struct dt_device *d)
 881 {
 882         CDEBUG(D_HA, "syncing OSD %s\n", LUSTRE_OSD_NAME);
 883         return ldiskfs_force_commit(osd_sb(osd_dt_dev(d)));
 884 }
 885
 886 /**
 887  * Start commit for OSD device.
 888  *
 889  * An implementation of dt_commit_async method for OSD device.
 890  * Asychronously starts underlayng fs sync and thereby a transaction
 891  * commit.
 892  *
 893  * \param env environment
 894  * \param d dt device
 895  *
 896  * \see dt_device_operations
 897  */
 898 static int osd_commit_async(const struct lu_env *env,
 899                             struct dt_device *d)
 900 {
 901         struct super_block *s = osd_sb(osd_dt_dev(d));
 902         ENTRY;
 903
 904         CDEBUG(D_HA, "async commit OSD %s\n", LUSTRE_OSD_NAME);
 905         RETURN(s->s_op->sync_fs(s, 0));
 906 }
 907
 908 /*
 909  * Concurrency: shouldn't matter.
 910  */
 911 lvfs_sbdev_type fsfilt_ldiskfs_journal_sbdev(struct super_block *);
 912
 913 static void osd_ro(const struct lu_env *env, struct dt_device *d)
 914 {
 915         ENTRY;
 916
 917         CERROR("*** setting device %s read-only ***\n", LUSTRE_OSD_NAME);
 918
 919         __lvfs_set_rdonly(lvfs_sbdev(osd_sb(osd_dt_dev(d))),
 920                           fsfilt_ldiskfs_journal_sbdev(osd_sb(osd_dt_dev(d))));
 921         EXIT;
 922 }
 923
 924
 925 /*
 926  * Concurrency: serialization provided by callers.
 927  */
 928 static int osd_init_capa_ctxt(const struct lu_env *env, struct dt_device *d,
 929                               int mode, unsigned long timeout, __u32 alg,
 930                               struct lustre_capa_key *keys)
 931 {
 932         struct osd_device *dev = osd_dt_dev(d);
 933         ENTRY;
 934
 935         dev->od_fl_capa = mode;
 936         dev->od_capa_timeout = timeout;
 937         dev->od_capa_alg = alg;
 938         dev->od_capa_keys = keys;
 939         RETURN(0);
 940 }
 941
 942 /**
 943  * Concurrency: serialization provided by callers.
 944  */
 945 static void osd_init_quota_ctxt(const struct lu_env *env, struct dt_device *d,
 946                                struct dt_quota_ctxt *ctxt, void *data)
 947 {
 948         struct obd_device *obd = (void *)ctxt;
 949         struct vfsmount *mnt = (struct vfsmount *)data;
 950         ENTRY;
 951
 952         obd->u.obt.obt_sb = mnt->mnt_root->d_inode->i_sb;
 953         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
 954         obd->obd_lvfs_ctxt.pwdmnt = mnt;
 955         obd->obd_lvfs_ctxt.pwd = mnt->mnt_root;
 956         obd->obd_lvfs_ctxt.fs = get_ds();
 957
 958         EXIT;
 959 }
 960
 961 /**
 962  * Note: we do not count into QUOTA here.
 963  * If we mount with --data_journal we may need more.
 964  */
 965 static const int osd_dto_credits_noquota[DTO_NR] = {
 966         /**
 967          * Insert/Delete.
 968          * INDEX_EXTRA_TRANS_BLOCKS(8) +
 969          * SINGLEDATA_TRANS_BLOCKS(8)
 970          * XXX Note: maybe iam need more, since iam have more level than
 971          *           EXT3 htree.
 972          */
 973         [DTO_INDEX_INSERT]  = 16,
 974         [DTO_INDEX_DELETE]  = 16,
 975         /**
 976          * Unused now
 977          */
 978         [DTO_IDNEX_UPDATE]  = 16,
 979         /**
 980          * Create a object. The same as create object in EXT3.
 981          * DATA_TRANS_BLOCKS(14) +
 982          * INDEX_EXTRA_BLOCKS(8) +
 983          * 3(inode bits, groups, GDT)
 984          */
 985         [DTO_OBJECT_CREATE] = 25,
 986         /**
 987          * Unused now
 988          */
 989         [DTO_OBJECT_DELETE] = 25,
 990         /**
 991          * Attr set credits.
 992          * 3(inode bits, group, GDT)
 993          */
 994         [DTO_ATTR_SET_BASE] = 3,
 995         /**
 996          * Xattr set. The same as xattr of EXT3.
 997          * DATA_TRANS_BLOCKS(14)
 998          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS
 999          * are also counted in. Do not know why?
1000          */
1001         [DTO_XATTR_SET]     = 14,
1002         [DTO_LOG_REC]       = 14,
1003         /**
1004          * creadits for inode change during write.
1005          */
1006         [DTO_WRITE_BASE]    = 3,
1007         /**
1008          * credits for single block write.
1009          */
1010         [DTO_WRITE_BLOCK]   = 14,
1011         /**
1012          * Attr set credits for chown.
1013          * This is extra credits for setattr, and it is null without quota
1014          */
1015         [DTO_ATTR_SET_CHOWN]= 0
1016 };
1017
1018 /**
1019  * Note: we count into QUOTA here.
1020  * If we mount with --data_journal we may need more.
1021  */
1022 static const int osd_dto_credits_quota[DTO_NR] = {
1023         /**
1024          * INDEX_EXTRA_TRANS_BLOCKS(8) +
1025          * SINGLEDATA_TRANS_BLOCKS(8) +
1026          * 2 * QUOTA_TRANS_BLOCKS(2)
1027          */
1028         [DTO_INDEX_INSERT]  = 20,
1029         /**
1030          * INDEX_EXTRA_TRANS_BLOCKS(8) +
1031          * SINGLEDATA_TRANS_BLOCKS(8) +
1032          * 2 * QUOTA_TRANS_BLOCKS(2)
1033          */
1034         [DTO_INDEX_DELETE]  = 20,
1035         /**
1036          * Unused now.
1037          */
1038         [DTO_IDNEX_UPDATE]  = 16,
1039         /*
1040          * Create a object. Same as create object in EXT3 filesystem.
1041          * DATA_TRANS_BLOCKS(16) +
1042          * INDEX_EXTRA_BLOCKS(8) +
1043          * 3(inode bits, groups, GDT) +
1044          * 2 * QUOTA_INIT_BLOCKS(25)
1045          */
1046         [DTO_OBJECT_CREATE] = 77,
1047         /*
1048          * Unused now.
1049          * DATA_TRANS_BLOCKS(16) +
1050          * INDEX_EXTRA_BLOCKS(8) +
1051          * 3(inode bits, groups, GDT) +
1052          * QUOTA(?)
1053          */
1054         [DTO_OBJECT_DELETE] = 27,
1055         /**
1056          * Attr set credits.
1057          * 3 (inode bit, group, GDT) +
1058          */
1059         [DTO_ATTR_SET_BASE] = 3,
1060         /**
1061          * Xattr set. The same as xattr of EXT3.
1062          * DATA_TRANS_BLOCKS(16)
1063          * XXX Note: in original MDS implmentation INDEX_EXTRA_TRANS_BLOCKS are
1064          *           also counted in. Do not know why?
1065          */
1066         [DTO_XATTR_SET]     = 16,
1067         [DTO_LOG_REC]       = 16,
1068         /**
1069          * creadits for inode change during write.
1070          */
1071         [DTO_WRITE_BASE]    = 3,
1072         /**
1073          * credits for single block write.
1074          */
1075         [DTO_WRITE_BLOCK]   = 16,
1076         /**
1077          * Attr set credits for chown.
1078          * It is added to already set setattr credits
1079          * 2 * QUOTA_INIT_BLOCKS(25) +
1080          * 2 * QUOTA_DEL_BLOCKS(9)
1081          */
1082         [DTO_ATTR_SET_CHOWN]= 68,
1083 };
1084
1085 static int osd_credit_get(const struct lu_env *env, struct dt_device *d,
1086                           enum dt_txn_op op)
1087 {
1088         LASSERT(ARRAY_SIZE(osd_dto_credits_noquota) ==
1089                 ARRAY_SIZE(osd_dto_credits_quota));
1090         LASSERT(0 <= op && op < ARRAY_SIZE(osd_dto_credits_noquota));
1091 #ifdef HAVE_QUOTA_SUPPORT
1092         if (test_opt(osd_sb(osd_dt_dev(d)), QUOTA))
1093                 return osd_dto_credits_quota[op];
1094         else
1095 #endif
1096                 return osd_dto_credits_noquota[op];
1097 }
1098
1099 static const struct dt_device_operations osd_dt_ops = {
1100         .dt_root_get       = osd_root_get,
1101         .dt_statfs         = osd_statfs,
1102         .dt_trans_start    = osd_trans_start,
1103         .dt_trans_stop     = osd_trans_stop,
1104         .dt_conf_get       = osd_conf_get,
1105         .dt_sync           = osd_sync,
1106         .dt_ro             = osd_ro,
1107         .dt_commit_async   = osd_commit_async,
1108         .dt_credit_get     = osd_credit_get,
1109         .dt_init_capa_ctxt = osd_init_capa_ctxt,
1110         .dt_init_quota_ctxt= osd_init_quota_ctxt,
1111 };
1112
1113 static void osd_object_read_lock(const struct lu_env *env,
1114                                  struct dt_object *dt, unsigned role)
1115 {
1116         struct osd_object *obj = osd_dt_obj(dt);
1117         struct osd_thread_info *oti = osd_oti_get(env);
1118
1119         LINVRNT(osd_invariant(obj));
1120
1121         LASSERT(obj->oo_owner != env);
1122         cfs_down_read_nested(&obj->oo_sem, role);
1123
1124         LASSERT(obj->oo_owner == NULL);
1125         oti->oti_r_locks++;
1126 }
1127
1128 static void osd_object_write_lock(const struct lu_env *env,
1129                                   struct dt_object *dt, unsigned role)
1130 {
1131         struct osd_object *obj = osd_dt_obj(dt);
1132         struct osd_thread_info *oti = osd_oti_get(env);
1133
1134         LINVRNT(osd_invariant(obj));
1135
1136         LASSERT(obj->oo_owner != env);
1137         cfs_down_write_nested(&obj->oo_sem, role);
1138
1139         LASSERT(obj->oo_owner == NULL);
1140         obj->oo_owner = env;
1141         oti->oti_w_locks++;
1142 }
1143
1144 static void osd_object_read_unlock(const struct lu_env *env,
1145                                    struct dt_object *dt)
1146 {
1147         struct osd_object *obj = osd_dt_obj(dt);
1148         struct osd_thread_info *oti = osd_oti_get(env);
1149
1150         LINVRNT(osd_invariant(obj));
1151
1152         LASSERT(oti->oti_r_locks > 0);
1153         oti->oti_r_locks--;
1154         cfs_up_read(&obj->oo_sem);
1155 }
1156
1157 static void osd_object_write_unlock(const struct lu_env *env,
1158                                     struct dt_object *dt)
1159 {
1160         struct osd_object *obj = osd_dt_obj(dt);
1161         struct osd_thread_info *oti = osd_oti_get(env);
1162
1163         LINVRNT(osd_invariant(obj));
1164
1165         LASSERT(obj->oo_owner == env);
1166         LASSERT(oti->oti_w_locks > 0);
1167         oti->oti_w_locks--;
1168         obj->oo_owner = NULL;
1169         cfs_up_write(&obj->oo_sem);
1170 }
1171
1172 static int osd_object_write_locked(const struct lu_env *env,
1173                                    struct dt_object *dt)
1174 {
1175         struct osd_object *obj = osd_dt_obj(dt);
1176
1177         LINVRNT(osd_invariant(obj));
1178
1179         return obj->oo_owner == env;
1180 }
1181
1182 static int capa_is_sane(const struct lu_env *env,
1183                         struct osd_device *dev,
1184                         struct lustre_capa *capa,
1185                         struct lustre_capa_key *keys)
1186 {
1187         struct osd_thread_info *oti = osd_oti_get(env);
1188         struct lustre_capa *tcapa = &oti->oti_capa;
1189         struct obd_capa *oc;
1190         int i, rc = 0;
1191         ENTRY;
1192
1193         oc = capa_lookup(dev->od_capa_hash, capa, 0);
1194         if (oc) {
1195                 if (capa_is_expired(oc)) {
1196                         DEBUG_CAPA(D_ERROR, capa, "expired");
1197                         rc = -ESTALE;
1198                 }
1199                 capa_put(oc);
1200                 RETURN(rc);
1201         }
1202
1203         if (capa_is_expired_sec(capa)) {
1204                 DEBUG_CAPA(D_ERROR, capa, "expired");
1205                 RETURN(-ESTALE);
1206         }
1207
1208         cfs_spin_lock(&capa_lock);
1209         for (i = 0; i < 2; i++) {
1210                 if (keys[i].lk_keyid == capa->lc_keyid) {
1211                         oti->oti_capa_key = keys[i];
1212                         break;
1213                 }
1214         }
1215         cfs_spin_unlock(&capa_lock);
1216
1217         if (i == 2) {
1218                 DEBUG_CAPA(D_ERROR, capa, "no matched capa key");
1219                 RETURN(-ESTALE);
1220         }
1221
1222         rc = capa_hmac(tcapa->lc_hmac, capa, oti->oti_capa_key.lk_key);
1223         if (rc)
1224                 RETURN(rc);
1225
1226         if (memcmp(tcapa->lc_hmac, capa->lc_hmac, sizeof(capa->lc_hmac))) {
1227                 DEBUG_CAPA(D_ERROR, capa, "HMAC mismatch");
1228                 RETURN(-EACCES);
1229         }
1230
1231         oc = capa_add(dev->od_capa_hash, capa);
1232         capa_put(oc);
1233
1234         RETURN(0);
1235 }
1236
1237 static int osd_object_auth(const struct lu_env *env, struct dt_object *dt,
1238                            struct lustre_capa *capa, __u64 opc)
1239 {
1240         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
1241         struct osd_device *dev = osd_dev(dt->do_lu.lo_dev);
1242         struct md_capainfo *ci;
1243         int rc;
1244
1245         if (!dev->od_fl_capa)
1246                 return 0;
1247
1248         if (capa == BYPASS_CAPA)
1249                 return 0;
1250
1251         ci = md_capainfo(env);
1252         if (unlikely(!ci))
1253                 return 0;
1254
1255         if (ci->mc_auth == LC_ID_NONE)
1256                 return 0;
1257
1258         if (!capa) {
1259                 CERROR("no capability is provided for fid "DFID"\n", PFID(fid));
1260                 return -EACCES;
1261         }
1262
1263         if (!lu_fid_eq(fid, &capa->lc_fid)) {
1264                 DEBUG_CAPA(D_ERROR, capa, "fid "DFID" mismatch with",
1265                            PFID(fid));
1266                 return -EACCES;
1267         }
1268
1269         if (!capa_opc_supported(capa, opc)) {
1270                 DEBUG_CAPA(D_ERROR, capa, "opc "LPX64" not supported by", opc);
1271                 return -EACCES;
1272         }
1273
1274         if ((rc = capa_is_sane(env, dev, capa, dev->od_capa_keys))) {
1275                 DEBUG_CAPA(D_ERROR, capa, "insane (rc %d)", rc);
1276                 return -EACCES;
1277         }
1278
1279         return 0;
1280 }
1281
1282 static struct timespec *osd_inode_time(const struct lu_env *env,
1283                                        struct inode *inode, __u64 seconds)
1284 {
1285         struct osd_thread_info *oti = osd_oti_get(env);
1286         struct timespec        *t   = &oti->oti_time;
1287
1288         t->tv_sec  = seconds;
1289         t->tv_nsec = 0;
1290         *t = timespec_trunc(*t, get_sb_time_gran(inode->i_sb));
1291         return t;
1292 }
1293
1294
1295 static void osd_inode_getattr(const struct lu_env *env,
1296                               struct inode *inode, struct lu_attr *attr)
1297 {
1298         attr->la_valid      |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE |
1299                                LA_SIZE | LA_BLOCKS | LA_UID | LA_GID |
1300                                LA_FLAGS | LA_NLINK | LA_RDEV | LA_BLKSIZE;
1301
1302         attr->la_atime      = LTIME_S(inode->i_atime);
1303         attr->la_mtime      = LTIME_S(inode->i_mtime);
1304         attr->la_ctime      = LTIME_S(inode->i_ctime);
1305         attr->la_mode       = inode->i_mode;
1306         attr->la_size       = i_size_read(inode);
1307         attr->la_blocks     = inode->i_blocks;
1308         attr->la_uid        = inode->i_uid;
1309         attr->la_gid        = inode->i_gid;
1310         attr->la_flags      = LDISKFS_I(inode)->i_flags;
1311         attr->la_nlink      = inode->i_nlink;
1312         attr->la_rdev       = inode->i_rdev;
1313         attr->la_blksize    = ll_inode_blksize(inode);
1314         attr->la_blkbits    = inode->i_blkbits;
1315 }
1316
1317 static int osd_attr_get(const struct lu_env *env,
1318                         struct dt_object *dt,
1319                         struct lu_attr *attr,
1320                         struct lustre_capa *capa)
1321 {
1322         struct osd_object *obj = osd_dt_obj(dt);
1323
1324         LASSERT(dt_object_exists(dt));
1325         LINVRNT(osd_invariant(obj));
1326
1327         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
1328                 return -EACCES;
1329
1330         cfs_spin_lock(&obj->oo_guard);
1331         osd_inode_getattr(env, obj->oo_inode, attr);
1332         cfs_spin_unlock(&obj->oo_guard);
1333         return 0;
1334 }
1335
1336 static int osd_inode_setattr(const struct lu_env *env,
1337                              struct inode *inode, const struct lu_attr *attr)
1338 {
1339         __u64 bits;
1340
1341         bits = attr->la_valid;
1342
1343         LASSERT(!(bits & LA_TYPE)); /* Huh? You want too much. */
1344
1345 #ifdef HAVE_QUOTA_SUPPORT
1346         if ((bits & LA_UID && attr->la_uid != inode->i_uid) ||
1347             (bits & LA_GID && attr->la_gid != inode->i_gid)) {
1348                 struct osd_ctxt *save = &osd_oti_get(env)->oti_ctxt;
1349                 struct iattr iattr;
1350                 int rc;
1351
1352                 iattr.ia_valid = 0;
1353                 if (bits & LA_UID)
1354                         iattr.ia_valid |= ATTR_UID;
1355                 if (bits & LA_GID)
1356                         iattr.ia_valid |= ATTR_GID;
1357                 iattr.ia_uid = attr->la_uid;
1358                 iattr.ia_gid = attr->la_gid;
1359                 osd_push_ctxt(env, save);
1360                 rc = DQUOT_TRANSFER(inode, &iattr) ? -EDQUOT : 0;
1361                 osd_pop_ctxt(save);
1362                 if (rc != 0)
1363                         return rc;
1364         }
1365 #endif
1366
1367         if (bits & LA_ATIME)
1368                 inode->i_atime  = *osd_inode_time(env, inode, attr->la_atime);
1369         if (bits & LA_CTIME)
1370                 inode->i_ctime  = *osd_inode_time(env, inode, attr->la_ctime);
1371         if (bits & LA_MTIME)
1372                 inode->i_mtime  = *osd_inode_time(env, inode, attr->la_mtime);
1373         if (bits & LA_SIZE) {
1374                 LDISKFS_I(inode)->i_disksize = attr->la_size;
1375                 i_size_write(inode, attr->la_size);
1376         }
1377
1378 #if 0
1379         /* OSD should not change "i_blocks" which is used by quota.
1380          * "i_blocks" should be changed by ldiskfs only. */
1381         if (bits & LA_BLOCKS)
1382                 inode->i_blocks = attr->la_blocks;
1383 #endif
1384         if (bits & LA_MODE)
1385                 inode->i_mode   = (inode->i_mode & S_IFMT) |
1386                         (attr->la_mode & ~S_IFMT);
1387         if (bits & LA_UID)
1388                 inode->i_uid    = attr->la_uid;
1389         if (bits & LA_GID)
1390                 inode->i_gid    = attr->la_gid;
1391         if (bits & LA_NLINK)
1392                 inode->i_nlink  = attr->la_nlink;
1393         if (bits & LA_RDEV)
1394                 inode->i_rdev   = attr->la_rdev;
1395
1396         if (bits & LA_FLAGS)
1397                 inode->i_flags = ll_ext_to_inode_flags(attr->la_flags);
1398         return 0;
1399 }
1400
1401 static int osd_attr_set(const struct lu_env *env,
1402                         struct dt_object *dt,
1403                         const struct lu_attr *attr,
1404                         struct thandle *handle,
1405                         struct lustre_capa *capa)
1406 {
1407         struct osd_object *obj = osd_dt_obj(dt);
1408         int rc;
1409
1410         LASSERT(handle != NULL);
1411         LASSERT(dt_object_exists(dt));
1412         LASSERT(osd_invariant(obj));
1413
1414         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
1415                 return -EACCES;
1416
1417         cfs_spin_lock(&obj->oo_guard);
1418         rc = osd_inode_setattr(env, obj->oo_inode, attr);
1419         cfs_spin_unlock(&obj->oo_guard);
1420
1421         if (!rc)
1422                 mark_inode_dirty(obj->oo_inode);
1423         return rc;
1424 }
1425
1426 /*
1427  * Object creation.
1428  *
1429  * XXX temporary solution.
1430  */
1431 static int osd_create_pre(struct osd_thread_info *info, struct osd_object *obj,
1432                           struct lu_attr *attr, struct thandle *th)
1433 {
1434         return 0;
1435 }
1436
1437 static int osd_create_post(struct osd_thread_info *info, struct osd_object *obj,
1438                            struct lu_attr *attr, struct thandle *th)
1439 {
1440         osd_object_init0(obj);
1441         return 0;
1442 }
1443
1444 static struct dentry * osd_child_dentry_get(const struct lu_env *env,
1445                                             struct osd_object *obj,
1446                                             const char *name,
1447                                             const int namelen)
1448 {
1449         struct osd_thread_info *info   = osd_oti_get(env);
1450         struct dentry *child_dentry = &info->oti_child_dentry;
1451         struct dentry *obj_dentry = &info->oti_obj_dentry;
1452
1453         obj_dentry->d_inode = obj->oo_inode;
1454         obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
1455         obj_dentry->d_name.hash = 0;
1456
1457         child_dentry->d_name.hash = 0;
1458         child_dentry->d_parent = obj_dentry;
1459         child_dentry->d_name.name = name;
1460         child_dentry->d_name.len = namelen;
1461         return child_dentry;
1462 }
1463
1464
1465 static int osd_mkfile(struct osd_thread_info *info, struct osd_object *obj,
1466                       cfs_umode_t mode,
1467                       struct dt_allocation_hint *hint,
1468                       struct thandle *th)
1469 {
1470         int result;
1471         struct osd_device  *osd = osd_obj2dev(obj);
1472         struct osd_thandle *oth;
1473         struct dt_object   *parent;
1474         struct inode       *inode;
1475 #ifdef HAVE_QUOTA_SUPPORT
1476         struct osd_ctxt    *save = &info->oti_ctxt;
1477 #endif
1478
1479         LINVRNT(osd_invariant(obj));
1480         LASSERT(obj->oo_inode == NULL);
1481
1482         oth = container_of(th, struct osd_thandle, ot_super);
1483         LASSERT(oth->ot_handle->h_transaction != NULL);
1484
1485         if (hint && hint->dah_parent)
1486                 parent = hint->dah_parent;
1487         else
1488                 parent = osd->od_obj_area;
1489
1490         LASSERT(parent != NULL);
1491         LASSERT(osd_dt_obj(parent)->oo_inode->i_op != NULL);
1492
1493 #ifdef HAVE_QUOTA_SUPPORT
1494         osd_push_ctxt(info->oti_env, save);
1495 #endif
1496         inode = ldiskfs_create_inode(oth->ot_handle,
1497                                      osd_dt_obj(parent)->oo_inode, mode);
1498 #ifdef HAVE_QUOTA_SUPPORT
1499         osd_pop_ctxt(save);
1500 #endif
1501         if (!IS_ERR(inode)) {
1502                 obj->oo_inode = inode;
1503                 result = 0;
1504         } else
1505                 result = PTR_ERR(inode);
1506         LINVRNT(osd_invariant(obj));
1507         return result;
1508 }
1509
1510 enum {
1511         OSD_NAME_LEN = 255
1512 };
1513
1514 static int osd_mkdir(struct osd_thread_info *info, struct osd_object *obj,
1515                      struct lu_attr *attr,
1516                      struct dt_allocation_hint *hint,
1517                      struct dt_object_format *dof,
1518                      struct thandle *th)
1519 {
1520         int result;
1521         struct osd_thandle *oth;
1522         struct osd_device *osd = osd_obj2dev(obj);
1523         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
1524
1525         LASSERT(S_ISDIR(attr->la_mode));
1526
1527         oth = container_of(th, struct osd_thandle, ot_super);
1528         LASSERT(oth->ot_handle->h_transaction != NULL);
1529         result = osd_mkfile(info, obj, mode, hint, th);
1530         if (result == 0 && osd->od_iop_mode == 0) {
1531                 LASSERT(obj->oo_inode != NULL);
1532                 /*
1533                  * XXX uh-oh... call low-level iam function directly.
1534                  */
1535
1536                 result = iam_lvar_create(obj->oo_inode, OSD_NAME_LEN, 4,
1537                                          sizeof (struct osd_fid_pack),
1538                                          oth->ot_handle);
1539         }
1540         return result;
1541 }
1542
1543 static int osd_mk_index(struct osd_thread_info *info, struct osd_object *obj,
1544                         struct lu_attr *attr,
1545                         struct dt_allocation_hint *hint,
1546                         struct dt_object_format *dof,
1547                         struct thandle *th)
1548 {
1549         int result;
1550         struct osd_thandle *oth;
1551         const struct dt_index_features *feat = dof->u.dof_idx.di_feat;
1552
1553         __u32 mode = (attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX));
1554
1555         LASSERT(S_ISREG(attr->la_mode));
1556
1557         oth = container_of(th, struct osd_thandle, ot_super);
1558         LASSERT(oth->ot_handle->h_transaction != NULL);
1559
1560         result = osd_mkfile(info, obj, mode, hint, th);
1561         if (result == 0) {
1562                 LASSERT(obj->oo_inode != NULL);
1563                 if (feat->dif_flags & DT_IND_VARKEY)
1564                         result = iam_lvar_create(obj->oo_inode,
1565                                                  feat->dif_keysize_max,
1566                                                  feat->dif_ptrsize,
1567                                                  feat->dif_recsize_max,
1568                                                  oth->ot_handle);
1569                 else
1570                         result = iam_lfix_create(obj->oo_inode,
1571                                                  feat->dif_keysize_max,
1572                                                  feat->dif_ptrsize,
1573                                                  feat->dif_recsize_max,
1574                                                  oth->ot_handle);
1575
1576         }
1577         return result;
1578 }
1579
1580 static int osd_mkreg(struct osd_thread_info *info, struct osd_object *obj,
1581                      struct lu_attr *attr,
1582                      struct dt_allocation_hint *hint,
1583                      struct dt_object_format *dof,
1584                      struct thandle *th)
1585 {
1586         LASSERT(S_ISREG(attr->la_mode));
1587         return osd_mkfile(info, obj, (attr->la_mode &
1588                                (S_IFMT | S_IRWXUGO | S_ISVTX)), hint, th);
1589 }
1590
1591 static int osd_mksym(struct osd_thread_info *info, struct osd_object *obj,
1592                      struct lu_attr *attr,
1593                      struct dt_allocation_hint *hint,
1594                      struct dt_object_format *dof,
1595                      struct thandle *th)
1596 {
1597         LASSERT(S_ISLNK(attr->la_mode));
1598         return osd_mkfile(info, obj, (attr->la_mode &
1599                               (S_IFMT | S_IRWXUGO | S_ISVTX)), hint, th);
1600 }
1601
1602 static int osd_mknod(struct osd_thread_info *info, struct osd_object *obj,
1603                      struct lu_attr *attr,
1604                      struct dt_allocation_hint *hint,
1605                      struct dt_object_format *dof,
1606                      struct thandle *th)
1607 {
1608         cfs_umode_t mode = attr->la_mode & (S_IFMT | S_IRWXUGO | S_ISVTX);
1609         int result;
1610
1611         LINVRNT(osd_invariant(obj));
1612         LASSERT(obj->oo_inode == NULL);
1613         LASSERT(S_ISCHR(mode) || S_ISBLK(mode) ||
1614                 S_ISFIFO(mode) || S_ISSOCK(mode));
1615
1616         result = osd_mkfile(info, obj, mode, hint, th);
1617         if (result == 0) {
1618                 LASSERT(obj->oo_inode != NULL);
1619                 init_special_inode(obj->oo_inode, mode, attr->la_rdev);
1620         }
1621         LINVRNT(osd_invariant(obj));
1622         return result;
1623 }
1624
1625 typedef int (*osd_obj_type_f)(struct osd_thread_info *, struct osd_object *,
1626                               struct lu_attr *,
1627                               struct dt_allocation_hint *hint,
1628                               struct dt_object_format *dof,
1629                               struct thandle *);
1630
1631 static osd_obj_type_f osd_create_type_f(enum dt_format_type type)
1632 {
1633         osd_obj_type_f result;
1634
1635         switch (type) {
1636         case DFT_DIR:
1637                 result = osd_mkdir;
1638                 break;
1639         case DFT_REGULAR:
1640                 result = osd_mkreg;
1641                 break;
1642         case DFT_SYM:
1643                 result = osd_mksym;
1644                 break;
1645         case DFT_NODE:
1646                 result = osd_mknod;
1647                 break;
1648         case DFT_INDEX:
1649                 result = osd_mk_index;
1650                 break;
1651
1652         default:
1653                 LBUG();
1654                 break;
1655         }
1656         return result;
1657 }
1658
1659
1660 static void osd_ah_init(const struct lu_env *env, struct dt_allocation_hint *ah,
1661                         struct dt_object *parent, cfs_umode_t child_mode)
1662 {
1663         LASSERT(ah);
1664
1665         memset(ah, 0, sizeof(*ah));
1666         ah->dah_parent = parent;
1667         ah->dah_mode = child_mode;
1668 }
1669
1670 /**
1671  * Helper function for osd_object_create()
1672  *
1673  * \retval 0, on success
1674  */
1675 static int __osd_object_create(struct osd_thread_info *info,
1676                                struct osd_object *obj, struct lu_attr *attr,
1677                                struct dt_allocation_hint *hint,
1678                                struct dt_object_format *dof,
1679                                struct thandle *th)
1680 {
1681
1682         int result;
1683
1684         result = osd_create_pre(info, obj, attr, th);
1685         if (result == 0) {
1686                 result = osd_create_type_f(dof->dof_type)(info, obj,
1687                                            attr, hint, dof, th);
1688                 if (result == 0)
1689                         result = osd_create_post(info, obj, attr, th);
1690         }
1691         return result;
1692 }
1693
1694 /**
1695  * Helper function for osd_object_create()
1696  *
1697  * \retval 0, on success
1698  */
1699 static int __osd_oi_insert(const struct lu_env *env, struct osd_object *obj,
1700                            const struct lu_fid *fid, struct thandle *th)
1701 {
1702         struct osd_thread_info *info = osd_oti_get(env);
1703         struct osd_inode_id    *id   = &info->oti_id;
1704         struct osd_device      *osd  = osd_obj2dev(obj);
1705         struct md_ucred        *uc   = md_ucred(env);
1706
1707         LASSERT(obj->oo_inode != NULL);
1708         LASSERT(uc != NULL);
1709
1710         id->oii_ino = obj->oo_inode->i_ino;
1711         id->oii_gen = obj->oo_inode->i_generation;
1712
1713         return osd_oi_insert(info, &osd->od_oi, fid, id, th,
1714                              uc->mu_cap & CFS_CAP_SYS_RESOURCE_MASK);
1715 }
1716
1717 static int osd_object_create(const struct lu_env *env, struct dt_object *dt,
1718                              struct lu_attr *attr,
1719                              struct dt_allocation_hint *hint,
1720                              struct dt_object_format *dof,
1721                              struct thandle *th)
1722 {
1723         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
1724         struct osd_object      *obj    = osd_dt_obj(dt);
1725         struct osd_thread_info *info   = osd_oti_get(env);
1726         int result;
1727
1728         ENTRY;
1729
1730         LINVRNT(osd_invariant(obj));
1731         LASSERT(!dt_object_exists(dt));
1732         LASSERT(osd_write_locked(env, obj));
1733         LASSERT(th != NULL);
1734
1735         result = __osd_object_create(info, obj, attr, hint, dof, th);
1736         if (result == 0)
1737                 result = __osd_oi_insert(env, obj, fid, th);
1738
1739         LASSERT(ergo(result == 0, dt_object_exists(dt)));
1740         LASSERT(osd_invariant(obj));
1741         RETURN(result);
1742 }
1743
1744 /**
1745  * Helper function for osd_xattr_set()
1746  */
1747 static int __osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
1748                            const struct lu_buf *buf, const char *name, int fl)
1749 {
1750         struct osd_object      *obj      = osd_dt_obj(dt);
1751         struct inode           *inode    = obj->oo_inode;
1752         struct osd_thread_info *info     = osd_oti_get(env);
1753         struct dentry          *dentry   = &info->oti_child_dentry;
1754         struct timespec        *t        = &info->oti_time;
1755         int                     fs_flags = 0;
1756         int  rc;
1757
1758         LASSERT(dt_object_exists(dt));
1759         LASSERT(inode->i_op != NULL && inode->i_op->setxattr != NULL);
1760         LASSERT(osd_write_locked(env, obj));
1761
1762         if (fl & LU_XATTR_REPLACE)
1763                 fs_flags |= XATTR_REPLACE;
1764
1765         if (fl & LU_XATTR_CREATE)
1766                 fs_flags |= XATTR_CREATE;
1767
1768         dentry->d_inode = inode;
1769         *t = inode->i_ctime;
1770         rc = inode->i_op->setxattr(dentry, name, buf->lb_buf,
1771                                    buf->lb_len, fs_flags);
1772         /* ctime should not be updated with server-side time. */
1773         cfs_spin_lock(&obj->oo_guard);
1774         inode->i_ctime = *t;
1775         cfs_spin_unlock(&obj->oo_guard);
1776         mark_inode_dirty(inode);
1777         return rc;
1778 }
1779
1780 /**
1781  * Put the fid into lustre_mdt_attrs, and then place the structure
1782  * inode's ea. This fid should not be altered during the life time
1783  * of the inode.
1784  *
1785  * \retval +ve, on success
1786  * \retval -ve, on error
1787  *
1788  * FIXME: It is good to have/use ldiskfs_xattr_set_handle() here
1789  */
1790 static int osd_ea_fid_set(const struct lu_env *env, struct dt_object *dt,
1791                           const struct lu_fid *fid)
1792 {
1793         struct osd_thread_info  *info      = osd_oti_get(env);
1794         struct lustre_mdt_attrs *mdt_attrs = &info->oti_mdt_attrs;
1795
1796         lustre_lma_init(mdt_attrs, fid);
1797         lustre_lma_swab(mdt_attrs);
1798         return __osd_xattr_set(env, dt,
1799                                osd_buf_get(env, mdt_attrs, sizeof *mdt_attrs),
1800                                XATTR_NAME_LMA, LU_XATTR_CREATE);
1801
1802 }
1803
1804 /**
1805  * Helper function to form igif
1806  */
1807 static inline void osd_igif_get(const struct lu_env *env, struct inode  *inode,
1808                                 struct lu_fid *fid)
1809 {
1810         LU_IGIF_BUILD(fid, inode->i_ino, inode->i_generation);
1811 }
1812
1813 /**
1814  * Helper function to pack the fid, ldiskfs stores fid in packed format.
1815  */
1816 void osd_fid_pack(struct osd_fid_pack *pack, const struct dt_rec *fid,
1817                   struct lu_fid *befider)
1818 {
1819         fid_cpu_to_be(befider, (struct lu_fid *)fid);
1820         memcpy(pack->fp_area, befider, sizeof(*befider));
1821         pack->fp_len =  sizeof(*befider) + 1;
1822 }
1823
1824 /**
1825  * ldiskfs supports fid in dirent, it is passed in dentry->d_fsdata.
1826  * lustre 1.8 also uses d_fsdata for passing other info to ldiskfs.
1827  * To have compatilibility with 1.8 ldiskfs driver we need to have
1828  * magic number at start of fid data.
1829  * \ldiskfs_dentry_param is used only to pass fid from osd to ldiskfs.
1830  * its inmemory API.
1831  */
1832 void osd_get_ldiskfs_dirent_param(struct ldiskfs_dentry_param *param,
1833                                   const struct dt_rec *fid)
1834 {
1835         param->edp_magic = LDISKFS_LUFID_MAGIC;
1836         param->edp_len =  sizeof(struct lu_fid) + 1;
1837
1838         fid_cpu_to_be((struct lu_fid *)param->edp_data,
1839                       (struct lu_fid *)fid);
1840 }
1841
1842 int osd_fid_unpack(struct lu_fid *fid, const struct osd_fid_pack *pack)
1843 {
1844         int result;
1845
1846         result = 0;
1847         switch (pack->fp_len) {
1848         case sizeof *fid + 1:
1849                 memcpy(fid, pack->fp_area, sizeof *fid);
1850                 fid_be_to_cpu(fid, fid);
1851                 break;
1852         default:
1853                 CERROR("Unexpected packed fid size: %d\n", pack->fp_len);
1854                 result = -EIO;
1855         }
1856         return result;
1857 }
1858
1859 /**
1860  * Try to read the fid from inode ea into dt_rec, if return value
1861  * i.e. rc is +ve, then we got fid, otherwise we will have to form igif
1862  *
1863  * \param fid object fid.
1864  *
1865  * \retval 0 on success
1866  */
1867 static int osd_ea_fid_get(const struct lu_env *env, struct osd_object *obj,
1868                           __u32 ino, struct lu_fid *fid)
1869 {
1870         struct osd_thread_info  *info      = osd_oti_get(env);
1871         struct lustre_mdt_attrs *mdt_attrs = &info->oti_mdt_attrs;
1872         struct lu_device        *ldev   = obj->oo_dt.do_lu.lo_dev;
1873         struct dentry           *dentry = &info->oti_child_dentry;
1874         struct osd_inode_id     *id     = &info->oti_id;
1875         struct osd_device       *dev;
1876         struct inode            *inode;
1877         int                      rc;
1878
1879         ENTRY;
1880         dev  = osd_dev(ldev);
1881
1882         id->oii_ino = ino;
1883         id->oii_gen = OSD_OII_NOGEN;
1884
1885         inode = osd_iget(info, dev, id);
1886         if (IS_ERR(inode)) {
1887                 rc = PTR_ERR(inode);
1888                 GOTO(out,rc);
1889         }
1890         dentry->d_inode = inode;
1891
1892         LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
1893         rc = inode->i_op->getxattr(dentry, XATTR_NAME_LMA, (void *)mdt_attrs,
1894                                    sizeof *mdt_attrs);
1895
1896         /* Check LMA compatibility */
1897         if (rc > 0 &&
1898             (mdt_attrs->lma_incompat & ~cpu_to_le32(LMA_INCOMPAT_SUPP))) {
1899                 CWARN("Inode %lx: Unsupported incompat LMA feature(s) %#x\n",
1900                       inode->i_ino, le32_to_cpu(mdt_attrs->lma_incompat) &
1901                       ~LMA_INCOMPAT_SUPP);
1902                 return -ENOSYS;
1903         }
1904
1905         if (rc > 0) {
1906                 lustre_lma_swab(mdt_attrs);
1907                 memcpy(fid, &mdt_attrs->lma_self_fid, sizeof(*fid));
1908                 rc = 0;
1909         } else if (rc == -ENODATA) {
1910                 osd_igif_get(env, inode, fid);
1911                 rc = 0;
1912         }
1913         iput(inode);
1914 out:
1915         RETURN(rc);
1916 }
1917
1918 /**
1919  * OSD layer object create function for interoperability mode (b11826).
1920  * This is mostly similar to osd_object_create(). Only difference being, fid is
1921  * inserted into inode ea here.
1922  *
1923  * \retval   0, on success
1924  * \retval -ve, on error
1925  */
1926 static int osd_object_ea_create(const struct lu_env *env, struct dt_object *dt,
1927                              struct lu_attr *attr,
1928                              struct dt_allocation_hint *hint,
1929                              struct dt_object_format *dof,
1930                              struct thandle *th)
1931 {
1932         const struct lu_fid    *fid    = lu_object_fid(&dt->do_lu);
1933         struct osd_object      *obj    = osd_dt_obj(dt);
1934         struct osd_thread_info *info   = osd_oti_get(env);
1935         int result;
1936
1937         ENTRY;
1938
1939         LASSERT(osd_invariant(obj));
1940         LASSERT(!dt_object_exists(dt));
1941         LASSERT(osd_write_locked(env, obj));
1942         LASSERT(th != NULL);
1943
1944         result = __osd_object_create(info, obj, attr, hint, dof, th);
1945
1946         /* objects under osd root shld have igif fid, so dont add fid EA */
1947         if (result == 0 && fid_seq(fid) >= FID_SEQ_NORMAL)
1948                 result = osd_ea_fid_set(env, dt, fid);
1949
1950         if (result == 0)
1951                 result = __osd_oi_insert(env, obj, fid, th);
1952
1953         LASSERT(ergo(result == 0, dt_object_exists(dt)));
1954         LINVRNT(osd_invariant(obj));
1955         RETURN(result);
1956 }
1957
1958 /*
1959  * Concurrency: @dt is write locked.
1960  */
1961 static void osd_object_ref_add(const struct lu_env *env,
1962                                struct dt_object *dt,
1963                                struct thandle *th)
1964 {
1965         struct osd_object *obj = osd_dt_obj(dt);
1966         struct inode *inode = obj->oo_inode;
1967
1968         LINVRNT(osd_invariant(obj));
1969         LASSERT(dt_object_exists(dt));
1970         LASSERT(osd_write_locked(env, obj));
1971         LASSERT(th != NULL);
1972
1973         cfs_spin_lock(&obj->oo_guard);
1974         LASSERT(inode->i_nlink < LDISKFS_LINK_MAX);
1975         inode->i_nlink++;
1976         cfs_spin_unlock(&obj->oo_guard);
1977         mark_inode_dirty(inode);
1978         LINVRNT(osd_invariant(obj));
1979 }
1980
1981 /*
1982  * Concurrency: @dt is write locked.
1983  */
1984 static void osd_object_ref_del(const struct lu_env *env,
1985                                struct dt_object *dt,
1986                                struct thandle *th)
1987 {
1988         struct osd_object *obj = osd_dt_obj(dt);
1989         struct inode *inode = obj->oo_inode;
1990
1991         LINVRNT(osd_invariant(obj));
1992         LASSERT(dt_object_exists(dt));
1993         LASSERT(osd_write_locked(env, obj));
1994         LASSERT(th != NULL);
1995
1996         cfs_spin_lock(&obj->oo_guard);
1997         LASSERT(inode->i_nlink > 0);
1998         inode->i_nlink--;
1999         cfs_spin_unlock(&obj->oo_guard);
2000         mark_inode_dirty(inode);
2001         LINVRNT(osd_invariant(obj));
2002 }
2003
2004 /*
2005  * Concurrency: @dt is read locked.
2006  */
2007 static int osd_xattr_get(const struct lu_env *env,
2008                          struct dt_object *dt,
2009                          struct lu_buf *buf,
2010                          const char *name,
2011                          struct lustre_capa *capa)
2012 {
2013         struct osd_object      *obj    = osd_dt_obj(dt);
2014         struct inode           *inode  = obj->oo_inode;
2015         struct osd_thread_info *info   = osd_oti_get(env);
2016         struct dentry          *dentry = &info->oti_obj_dentry;
2017
2018         LASSERT(dt_object_exists(dt));
2019         LASSERT(inode->i_op != NULL && inode->i_op->getxattr != NULL);
2020         LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
2021
2022         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2023                 return -EACCES;
2024
2025         dentry->d_inode = inode;
2026         return inode->i_op->getxattr(dentry, name, buf->lb_buf, buf->lb_len);
2027 }
2028
2029 /*
2030  * Concurrency: @dt is write locked.
2031  */
2032 static int osd_xattr_set(const struct lu_env *env, struct dt_object *dt,
2033                          const struct lu_buf *buf, const char *name, int fl,
2034                          struct thandle *handle, struct lustre_capa *capa)
2035 {
2036         LASSERT(handle != NULL);
2037
2038         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2039                 return -EACCES;
2040
2041         return __osd_xattr_set(env, dt, buf, name, fl);
2042 }
2043
2044 /*
2045  * Concurrency: @dt is read locked.
2046  */
2047 static int osd_xattr_list(const struct lu_env *env,
2048                           struct dt_object *dt,
2049                           struct lu_buf *buf,
2050                           struct lustre_capa *capa)
2051 {
2052         struct osd_object      *obj    = osd_dt_obj(dt);
2053         struct inode           *inode  = obj->oo_inode;
2054         struct osd_thread_info *info   = osd_oti_get(env);
2055         struct dentry          *dentry = &info->oti_obj_dentry;
2056
2057         LASSERT(dt_object_exists(dt));
2058         LASSERT(inode->i_op != NULL && inode->i_op->listxattr != NULL);
2059         LASSERT(osd_read_locked(env, obj) || osd_write_locked(env, obj));
2060
2061         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_READ))
2062                 return -EACCES;
2063
2064         dentry->d_inode = inode;
2065         return inode->i_op->listxattr(dentry, buf->lb_buf, buf->lb_len);
2066 }
2067
2068 /*
2069  * Concurrency: @dt is write locked.
2070  */
2071 static int osd_xattr_del(const struct lu_env *env,
2072                          struct dt_object *dt,
2073                          const char *name,
2074                          struct thandle *handle,
2075                          struct lustre_capa *capa)
2076 {
2077         struct osd_object      *obj    = osd_dt_obj(dt);
2078         struct inode           *inode  = obj->oo_inode;
2079         struct osd_thread_info *info   = osd_oti_get(env);
2080         struct dentry          *dentry = &info->oti_obj_dentry;
2081         struct timespec        *t      = &info->oti_time;
2082         int                     rc;
2083
2084         LASSERT(dt_object_exists(dt));
2085         LASSERT(inode->i_op != NULL && inode->i_op->removexattr != NULL);
2086         LASSERT(osd_write_locked(env, obj));
2087         LASSERT(handle != NULL);
2088
2089         if (osd_object_auth(env, dt, capa, CAPA_OPC_META_WRITE))
2090                 return -EACCES;
2091
2092         dentry->d_inode = inode;
2093         *t = inode->i_ctime;
2094         rc = inode->i_op->removexattr(dentry, name);
2095         /* ctime should not be updated with server-side time. */
2096         cfs_spin_lock(&obj->oo_guard);
2097         inode->i_ctime = *t;
2098         cfs_spin_unlock(&obj->oo_guard);
2099         mark_inode_dirty(inode);
2100         return rc;
2101 }
2102
2103 static struct obd_capa *osd_capa_get(const struct lu_env *env,
2104                                      struct dt_object *dt,
2105                                      struct lustre_capa *old,
2106                                      __u64 opc)
2107 {
2108         struct osd_thread_info *info = osd_oti_get(env);
2109         const struct lu_fid *fid = lu_object_fid(&dt->do_lu);
2110         struct osd_object *obj = osd_dt_obj(dt);
2111         struct osd_device *dev = osd_obj2dev(obj);
2112         struct lustre_capa_key *key = &info->oti_capa_key;
2113         struct lustre_capa *capa = &info->oti_capa;
2114         struct obd_capa *oc;
2115         struct md_capainfo *ci;
2116         int rc;
2117         ENTRY;
2118
2119         if (!dev->od_fl_capa)
2120                 RETURN(ERR_PTR(-ENOENT));
2121
2122         LASSERT(dt_object_exists(dt));
2123         LINVRNT(osd_invariant(obj));
2124
2125         /* renewal sanity check */
2126         if (old && osd_object_auth(env, dt, old, opc))
2127                 RETURN(ERR_PTR(-EACCES));
2128
2129         ci = md_capainfo(env);
2130         if (unlikely(!ci))
2131                 RETURN(ERR_PTR(-ENOENT));
2132
2133         switch (ci->mc_auth) {
2134         case LC_ID_NONE:
2135                 RETURN(NULL);
2136         case LC_ID_PLAIN:
2137                 capa->lc_uid = obj->oo_inode->i_uid;
2138                 capa->lc_gid = obj->oo_inode->i_gid;
2139                 capa->lc_flags = LC_ID_PLAIN;
2140                 break;
2141         case LC_ID_CONVERT: {
2142                 __u32 d[4], s[4];
2143
2144                 s[0] = obj->oo_inode->i_uid;
2145                 cfs_get_random_bytes(&(s[1]), sizeof(__u32));
2146                 s[2] = obj->oo_inode->i_gid;
2147                 cfs_get_random_bytes(&(s[3]), sizeof(__u32));
2148                 rc = capa_encrypt_id(d, s, key->lk_key, CAPA_HMAC_KEY_MAX_LEN);
2149                 if (unlikely(rc))
2150                         RETURN(ERR_PTR(rc));
2151
2152                 capa->lc_uid   = ((__u64)d[1] << 32) | d[0];
2153                 capa->lc_gid   = ((__u64)d[3] << 32) | d[2];
2154                 capa->lc_flags = LC_ID_CONVERT;
2155                 break;
2156         }
2157         default:
2158                 RETURN(ERR_PTR(-EINVAL));
2159         }
2160
2161         capa->lc_fid = *fid;
2162         capa->lc_opc = opc;
2163         capa->lc_flags |= dev->od_capa_alg << 24;
2164         capa->lc_timeout = dev->od_capa_timeout;
2165         capa->lc_expiry = 0;
2166
2167         oc = capa_lookup(dev->od_capa_hash, capa, 1);
2168         if (oc) {
2169                 LASSERT(!capa_is_expired(oc));
2170                 RETURN(oc);
2171         }
2172
2173         cfs_spin_lock(&capa_lock);
2174         *key = dev->od_capa_keys[1];
2175         cfs_spin_unlock(&capa_lock);
2176
2177         capa->lc_keyid = key->lk_keyid;
2178         capa->lc_expiry = cfs_time_current_sec() + dev->od_capa_timeout;
2179
2180         rc = capa_hmac(capa->lc_hmac, capa, key->lk_key);
2181         if (rc) {
2182                 DEBUG_CAPA(D_ERROR, capa, "HMAC failed: %d for", rc);
2183                 RETURN(ERR_PTR(rc));
2184         }
2185
2186         oc = capa_add(dev->od_capa_hash, capa);
2187         RETURN(oc);
2188 }
2189
2190 static int osd_object_sync(const struct lu_env *env, struct dt_object *dt)
2191 {
2192         int rc;
2193         struct osd_object      *obj    = osd_dt_obj(dt);
2194         struct inode           *inode  = obj->oo_inode;
2195         struct osd_thread_info *info   = osd_oti_get(env);
2196         struct dentry          *dentry = &info->oti_obj_dentry;
2197         struct file            *file   = &info->oti_file;
2198         ENTRY;
2199
2200         dentry->d_inode = inode;
2201         file->f_dentry = dentry;
2202         file->f_mapping = inode->i_mapping;
2203         file->f_op = inode->i_fop;
2204         LOCK_INODE_MUTEX(inode);
2205         rc = file->f_op->fsync(file, dentry, 0);
2206         UNLOCK_INODE_MUTEX(inode);
2207         RETURN(rc);
2208 }
2209
2210 /*
2211  * Get the 64-bit version for an inode.
2212  */
2213 static dt_obj_version_t osd_object_version_get(const struct lu_env *env,
2214                                                struct dt_object *dt)
2215 {
2216         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2217
2218         CDEBUG(D_INFO, "Get version "LPX64" for inode %lu\n",
2219                LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2220         return LDISKFS_I(inode)->i_fs_version;
2221 }
2222
2223 /*
2224  * Set the 64-bit version and return the old version.
2225  */
2226 static void osd_object_version_set(const struct lu_env *env, struct dt_object *dt,
2227                                    dt_obj_version_t new_version)
2228 {
2229         struct inode *inode = osd_dt_obj(dt)->oo_inode;
2230
2231         CDEBUG(D_INFO, "Set version "LPX64" (old "LPX64") for inode %lu\n",
2232                new_version, LDISKFS_I(inode)->i_fs_version, inode->i_ino);
2233         LDISKFS_I(inode)->i_fs_version = new_version;
2234         /** Version is set after all inode operations are finished,
2235          *  so we should mark it dirty here */
2236         inode->i_sb->s_op->dirty_inode(inode);
2237 }
2238
2239 static int osd_data_get(const struct lu_env *env, struct dt_object *dt,
2240                         void **data)
2241 {
2242         struct osd_object *obj = osd_dt_obj(dt);
2243         ENTRY;
2244
2245         *data = (void *)obj->oo_inode;
2246         RETURN(0);
2247 }
2248
2249 /*
2250  * Index operations.
2251  */
2252
2253 static int osd_iam_index_probe(const struct lu_env *env, struct osd_object *o,
2254                            const struct dt_index_features *feat)
2255 {
2256         struct iam_descr *descr;
2257
2258         if (osd_object_is_root(o))
2259                 return feat == &dt_directory_features;
2260
2261         LASSERT(o->oo_dir != NULL);
2262
2263         descr = o->oo_dir->od_container.ic_descr;
2264         if (feat == &dt_directory_features) {
2265                 if (descr->id_rec_size == sizeof(struct osd_fid_pack))
2266                         return 1;
2267                 else
2268                         return 0;
2269         } else {
2270                 return
2271                         feat->dif_keysize_min <= descr->id_key_size &&
2272                         descr->id_key_size <= feat->dif_keysize_max &&
2273                         feat->dif_recsize_min <= descr->id_rec_size &&
2274                         descr->id_rec_size <= feat->dif_recsize_max &&
2275                         !(feat->dif_flags & (DT_IND_VARKEY |
2276                                              DT_IND_VARREC | DT_IND_NONUNQ)) &&
2277                         ergo(feat->dif_flags & DT_IND_UPDATE,
2278                              1 /* XXX check that object (and file system) is
2279                                 * writable */);
2280         }
2281 }
2282
2283 static int osd_iam_container_init(const struct lu_env *env,
2284                                   struct osd_object *obj,
2285                                   struct osd_directory *dir)
2286 {
2287         int result;
2288         struct iam_container *bag;
2289
2290         bag    = &dir->od_container;
2291         result = iam_container_init(bag, &dir->od_descr, obj->oo_inode);
2292         if (result == 0) {
2293                 result = iam_container_setup(bag);
2294                 if (result == 0)
2295                         obj->oo_dt.do_index_ops = &osd_index_iam_ops;
2296                 else
2297                         iam_container_fini(bag);
2298         }
2299         return result;
2300 }
2301
2302
2303 /*
2304  * Concurrency: no external locking is necessary.
2305  */
2306 static int osd_index_try(const struct lu_env *env, struct dt_object *dt,
2307                          const struct dt_index_features *feat)
2308 {
2309         int result;
2310         int ea_dir = 0;
2311         struct osd_object *obj = osd_dt_obj(dt);
2312         struct osd_device *osd = osd_obj2dev(obj);
2313
2314         LINVRNT(osd_invariant(obj));
2315         LASSERT(dt_object_exists(dt));
2316
2317         if (osd_object_is_root(obj)) {
2318                 dt->do_index_ops = &osd_index_ea_ops;
2319                 result = 0;
2320         } else if (feat == &dt_directory_features && osd->od_iop_mode) {
2321                 dt->do_index_ops = &osd_index_ea_ops;
2322                 if (S_ISDIR(obj->oo_inode->i_mode))
2323                         result = 0;
2324                 else
2325                         result = -ENOTDIR;
2326                 ea_dir = 1;
2327         } else if (!osd_has_index(obj)) {
2328                 struct osd_directory *dir;
2329
2330                 OBD_ALLOC_PTR(dir);
2331                 if (dir != NULL) {
2332
2333                         cfs_spin_lock(&obj->oo_guard);
2334                         if (obj->oo_dir == NULL)
2335                                 obj->oo_dir = dir;
2336                         else
2337                                 /*
2338                                  * Concurrent thread allocated container data.
2339                                  */
2340                                 OBD_FREE_PTR(dir);
2341                         cfs_spin_unlock(&obj->oo_guard);
2342                         /*
2343                          * Now, that we have container data, serialize its
2344                          * initialization.
2345                          */
2346                         cfs_down_write(&obj->oo_ext_idx_sem);
2347                         /*
2348                          * recheck under lock.
2349                          */
2350                         if (!osd_has_index(obj))
2351                                 result = osd_iam_container_init(env, obj, dir);
2352                         else
2353                                 result = 0;
2354                         cfs_up_write(&obj->oo_ext_idx_sem);
2355                 } else
2356                         result = -ENOMEM;
2357         } else
2358                 result = 0;
2359
2360         if (result == 0 && ea_dir == 0) {
2361                 if (!osd_iam_index_probe(env, obj, feat))
2362                         result = -ENOTDIR;
2363         }
2364         LINVRNT(osd_invariant(obj));
2365
2366         return result;
2367 }
2368
2369 static const struct dt_object_operations osd_obj_ops = {
2370         .do_read_lock    = osd_object_read_lock,
2371         .do_write_lock   = osd_object_write_lock,
2372         .do_read_unlock  = osd_object_read_unlock,
2373         .do_write_unlock = osd_object_write_unlock,
2374         .do_write_locked = osd_object_write_locked,
2375         .do_attr_get     = osd_attr_get,
2376         .do_attr_set     = osd_attr_set,
2377         .do_ah_init      = osd_ah_init,
2378         .do_create       = osd_object_create,
2379         .do_index_try    = osd_index_try,
2380         .do_ref_add      = osd_object_ref_add,
2381         .do_ref_del      = osd_object_ref_del,
2382         .do_xattr_get    = osd_xattr_get,
2383         .do_xattr_set    = osd_xattr_set,
2384         .do_xattr_del    = osd_xattr_del,
2385         .do_xattr_list   = osd_xattr_list,
2386         .do_capa_get     = osd_capa_get,
2387         .do_object_sync  = osd_object_sync,
2388         .do_version_get  = osd_object_version_get,
2389         .do_version_set  = osd_object_version_set,
2390         .do_data_get     = osd_data_get,
2391 };
2392
2393 /**
2394  * dt_object_operations for interoperability mode
2395  * (i.e. to run 2.0 mds on 1.8 disk) (b11826)
2396  */
2397 static const struct dt_object_operations osd_obj_ea_ops = {
2398         .do_read_lock    = osd_object_read_lock,
2399         .do_write_lock   = osd_object_write_lock,
2400         .do_read_unlock  = osd_object_read_unlock,
2401         .do_write_unlock = osd_object_write_unlock,
2402         .do_write_locked = osd_object_write_locked,
2403         .do_attr_get     = osd_attr_get,
2404         .do_attr_set     = osd_attr_set,
2405         .do_ah_init      = osd_ah_init,
2406         .do_create       = osd_object_ea_create,
2407         .do_index_try    = osd_index_try,
2408         .do_ref_add      = osd_object_ref_add,
2409         .do_ref_del      = osd_object_ref_del,
2410         .do_xattr_get    = osd_xattr_get,
2411         .do_xattr_set    = osd_xattr_set,
2412         .do_xattr_del    = osd_xattr_del,
2413         .do_xattr_list   = osd_xattr_list,
2414         .do_capa_get     = osd_capa_get,
2415         .do_object_sync  = osd_object_sync,
2416         .do_version_get  = osd_object_version_get,
2417         .do_version_set  = osd_object_version_set,
2418         .do_data_get     = osd_data_get,
2419 };
2420
2421 /*
2422  * Body operations.
2423  */
2424
2425 /*
2426  * XXX: Another layering violation for now.
2427  *
2428  * We don't want to use ->f_op->read methods, because generic file write
2429  *
2430  *         - serializes on ->i_sem, and
2431  *
2432  *         - does a lot of extra work like balance_dirty_pages(),
2433  *
2434  * which doesn't work for globally shared files like /last-received.
2435  */
2436 static int osd_ldiskfs_readlink(struct inode *inode, char *buffer, int buflen)
2437 {
2438         struct ldiskfs_inode_info *ei = LDISKFS_I(inode);
2439
2440         memcpy(buffer, (char*)ei->i_data, buflen);
2441
2442         return  buflen;
2443 }
2444
2445 static int osd_ldiskfs_read(struct inode *inode, void *buf, int size,
2446                             loff_t *offs)
2447 {
2448         struct buffer_head *bh;
2449         unsigned long block;
2450         int osize = size;
2451         int blocksize;
2452         int csize;
2453         int boffs;
2454         int err;
2455
2456         /* prevent reading after eof */
2457         spin_lock(&inode->i_lock);
2458         if (i_size_read(inode) < *offs + size) {
2459                 size = i_size_read(inode) - *offs;
2460                 spin_unlock(&inode->i_lock);
2461                 if (size < 0) {
2462                         CDEBUG(D_EXT2, "size %llu is too short to read @%llu\n",
2463                                i_size_read(inode), *offs);
2464                         return -EBADR;
2465                 } else if (size == 0) {
2466                         return 0;
2467                 }
2468         } else {
2469                 spin_unlock(&inode->i_lock);
2470         }
2471
2472         blocksize = 1 << inode->i_blkbits;
2473
2474         while (size > 0) {
2475                 block = *offs >> inode->i_blkbits;
2476                 boffs = *offs & (blocksize - 1);
2477                 csize = min(blocksize - boffs, size);
2478                 bh = ldiskfs_bread(NULL, inode, block, 0, &err);
2479                 if (!bh) {
2480                         CERROR("can't read block: %d\n", err);
2481                         return err;
2482                 }
2483
2484                 memcpy(buf, bh->b_data + boffs, csize);
2485                 brelse(bh);
2486
2487                 *offs += csize;
2488                 buf += csize;
2489                 size -= csize;
2490         }
2491         return osize;
2492 }
2493
2494 static ssize_t osd_read(const struct lu_env *env, struct dt_object *dt,
2495                         struct lu_buf *buf, loff_t *pos,
2496                         struct lustre_capa *capa)
2497 {
2498         struct osd_object      *obj    = osd_dt_obj(dt);
2499         struct inode           *inode  = obj->oo_inode;
2500         int rc;
2501
2502         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
2503                 RETURN(-EACCES);
2504
2505         /* Read small symlink from inode body as we need to maintain correct
2506          * on-disk symlinks for ldiskfs.
2507          */
2508         if (S_ISLNK(obj->oo_dt.do_lu.lo_header->loh_attr) &&
2509             (buf->lb_len <= sizeof (LDISKFS_I(inode)->i_data)))
2510                 rc = osd_ldiskfs_readlink(inode, buf->lb_buf, buf->lb_len);
2511         else
2512                 rc = osd_ldiskfs_read(inode, buf->lb_buf, buf->lb_len, pos);
2513
2514         return rc;
2515 }
2516
2517 static int osd_ldiskfs_writelink(struct inode *inode, char *buffer, int buflen)
2518 {
2519
2520         memcpy((char*)&LDISKFS_I(inode)->i_data, (char *)buffer,
2521                buflen);
2522         LDISKFS_I(inode)->i_disksize = buflen;
2523         i_size_write(inode, buflen);
2524         inode->i_sb->s_op->dirty_inode(inode);
2525
2526         return 0;
2527 }
2528
2529 static int osd_ldiskfs_write_record(struct inode *inode, void *buf, int bufsize,
2530                                     loff_t *offs, handle_t *handle)
2531 {
2532         struct buffer_head *bh = NULL;
2533         loff_t offset = *offs;
2534         loff_t new_size = i_size_read(inode);
2535         unsigned long block;
2536         int blocksize = 1 << inode->i_blkbits;
2537         int err = 0;
2538         int size;
2539         int boffs;
2540         int dirty_inode = 0;
2541
2542         while (bufsize > 0) {
2543                 if (bh != NULL)
2544                         brelse(bh);
2545
2546                 block = offset >> inode->i_blkbits;
2547                 boffs = offset & (blocksize - 1);
2548                 size = min(blocksize - boffs, bufsize);
2549                 bh = ldiskfs_bread(handle, inode, block, 1, &err);
2550                 if (!bh) {
2551                         CERROR("can't read/create block: %d\n", err);
2552                         break;
2553                 }
2554
2555                 err = ldiskfs_journal_get_write_access(handle, bh);
2556                 if (err) {
2557                         CERROR("journal_get_write_access() returned error %d\n",
2558                                err);
2559                         break;
2560                 }
2561                 LASSERTF(boffs + size <= bh->b_size,
2562                          "boffs %d size %d bh->b_size %lu",
2563                          boffs, size, (unsigned long)bh->b_size);
2564                 memcpy(bh->b_data + boffs, buf, size);
2565                 err = ldiskfs_journal_dirty_metadata(handle, bh);
2566                 if (err)
2567                         break;
2568
2569                 if (offset + size > new_size)
2570                         new_size = offset + size;
2571                 offset += size;
2572                 bufsize -= size;
2573                 buf += size;
2574         }
2575         if (bh)
2576                 brelse(bh);
2577
2578         /* correct in-core and on-disk sizes */
2579         if (new_size > i_size_read(inode)) {
2580                 spin_lock(&inode->i_lock);
2581                 if (new_size > i_size_read(inode))
2582                         i_size_write(inode, new_size);
2583                 if (i_size_read(inode) > LDISKFS_I(inode)->i_disksize) {
2584                         LDISKFS_I(inode)->i_disksize = i_size_read(inode);
2585                         dirty_inode = 1;
2586                 }
2587                 spin_unlock(&inode->i_lock);
2588                 if (dirty_inode)
2589                         inode->i_sb->s_op->dirty_inode(inode);
2590         }
2591
2592         if (err == 0)
2593                 *offs = offset;
2594         return err;
2595 }
2596
2597 static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
2598                          const struct lu_buf *buf, loff_t *pos,
2599                          struct thandle *handle, struct lustre_capa *capa,
2600                          int ignore_quota)
2601 {
2602         struct osd_object  *obj   = osd_dt_obj(dt);
2603         struct inode       *inode = obj->oo_inode;
2604         struct osd_thandle *oh;
2605         ssize_t            result = 0;
2606 #ifdef HAVE_QUOTA_SUPPORT
2607         cfs_cap_t           save = current->cap_effective;
2608 #endif
2609
2610         LASSERT(handle != NULL);
2611
2612         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_WRITE))
2613                 RETURN(-EACCES);
2614
2615         oh = container_of(handle, struct osd_thandle, ot_super);
2616         LASSERT(oh->ot_handle->h_transaction != NULL);
2617 #ifdef HAVE_QUOTA_SUPPORT
2618         if (ignore_quota)
2619                 current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
2620         else
2621                 current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
2622 #endif
2623         /* Write small symlink to inode body as we need to maintain correct
2624          * on-disk symlinks for ldiskfs.
2625          */
2626         if(S_ISLNK(obj->oo_dt.do_lu.lo_header->loh_attr) &&
2627            (buf->lb_len < sizeof (LDISKFS_I(inode)->i_data)))
2628                 result = osd_ldiskfs_writelink(inode, buf->lb_buf, buf->lb_len);
2629         else
2630                 result = osd_ldiskfs_write_record(inode, buf->lb_buf,
2631                                                   buf->lb_len, pos,
2632                                                   oh->ot_handle);
2633 #ifdef HAVE_QUOTA_SUPPORT
2634         current->cap_effective = save;
2635 #endif
2636         if (result == 0)
2637                 result = buf->lb_len;
2638         return result;
2639 }
2640
2641 static const struct dt_body_operations osd_body_ops = {
2642         .dbo_read  = osd_read,
2643         .dbo_write = osd_write
2644 };
2645
2646
2647 /**
2648  *      delete a (key, value) pair from index \a dt specified by \a key
2649  *
2650  *      \param  dt      osd index object
2651  *      \param  key     key for index
2652  *      \param  rec     record reference
2653  *      \param  handle  transaction handler
2654  *
2655  *      \retval  0  success
2656  *      \retval -ve   failure
2657  */
2658
2659 static int osd_index_iam_delete(const struct lu_env *env, struct dt_object *dt,
2660                                 const struct dt_key *key, struct thandle *handle,
2661                                 struct lustre_capa *capa)
2662 {
2663         struct osd_object     *obj = osd_dt_obj(dt);
2664         struct osd_thandle    *oh;
2665         struct iam_path_descr *ipd;
2666         struct iam_container  *bag = &obj->oo_dir->od_container;
2667         int rc;
2668
2669         ENTRY;
2670
2671         LINVRNT(osd_invariant(obj));
2672         LASSERT(dt_object_exists(dt));
2673         LASSERT(bag->ic_object == obj->oo_inode);
2674         LASSERT(handle != NULL);
2675
2676         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
2677                 RETURN(-EACCES);
2678
2679         ipd = osd_idx_ipd_get(env, bag);
2680         if (unlikely(ipd == NULL))
2681                 RETURN(-ENOMEM);
2682
2683         oh = container_of0(handle, struct osd_thandle, ot_super);
2684         LASSERT(oh->ot_handle != NULL);
2685         LASSERT(oh->ot_handle->h_transaction != NULL);
2686
2687         rc = iam_delete(oh->ot_handle, bag, (const struct iam_key *)key, ipd);
2688         osd_ipd_put(env, bag, ipd);
2689         LINVRNT(osd_invariant(obj));
2690         RETURN(rc);
2691 }
2692
2693 static inline int osd_get_fid_from_dentry(struct ldiskfs_dir_entry_2 *de,
2694                                           struct dt_rec *fid)
2695 {
2696         struct osd_fid_pack *rec;
2697         int rc = -ENODATA;
2698
2699         if (de->file_type & LDISKFS_DIRENT_LUFID) {
2700                 rec = (struct osd_fid_pack *) (de->name + de->name_len + 1);
2701                 rc = osd_fid_unpack((struct lu_fid *)fid, rec);
2702         }
2703         RETURN(rc);
2704 }
2705
2706 /**
2707  * Index delete function for interoperability mode (b11826).
2708  * It will remove the directory entry added by osd_index_ea_insert().
2709  * This entry is needed to maintain name->fid mapping.
2710  *
2711  * \param key,  key i.e. file entry to be deleted
2712  *
2713  * \retval   0, on success
2714  * \retval -ve, on error
2715  */
2716 static int osd_index_ea_delete(const struct lu_env *env, struct dt_object *dt,
2717                                const struct dt_key *key, struct thandle *handle,
2718                                struct lustre_capa *capa)
2719 {
2720         struct osd_object          *obj    = osd_dt_obj(dt);
2721         struct inode               *dir    = obj->oo_inode;
2722         struct dentry              *dentry;
2723         struct osd_thandle         *oh;
2724         struct ldiskfs_dir_entry_2 *de;
2725         struct buffer_head         *bh;
2726
2727         int rc;
2728
2729         ENTRY;
2730
2731         LINVRNT(osd_invariant(obj));
2732         LASSERT(dt_object_exists(dt));
2733         LASSERT(handle != NULL);
2734
2735         oh = container_of(handle, struct osd_thandle, ot_super);
2736         LASSERT(oh->ot_handle != NULL);
2737         LASSERT(oh->ot_handle->h_transaction != NULL);
2738
2739         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_DELETE))
2740                 RETURN(-EACCES);
2741
2742         dentry = osd_child_dentry_get(env, obj,
2743                                       (char *)key, strlen((char *)key));
2744
2745         cfs_down_write(&obj->oo_ext_idx_sem);
2746         bh = ll_ldiskfs_find_entry(dir, dentry, &de);
2747         if (bh) {
2748                 struct osd_thread_info *oti = osd_oti_get(env);
2749                 struct timespec *ctime = &oti->oti_time;
2750                 struct timespec *mtime = &oti->oti_time2;
2751
2752                 *ctime = dir->i_ctime;
2753                 *mtime = dir->i_mtime;
2754                 rc = ldiskfs_delete_entry(oh->ot_handle,
2755                                 dir, de, bh);
2756                 /* xtime should not be updated with server-side time. */
2757                 cfs_spin_lock(&obj->oo_guard);
2758                 dir->i_ctime = *ctime;
2759                 dir->i_mtime = *mtime;
2760                 cfs_spin_unlock(&obj->oo_guard);
2761                 mark_inode_dirty(dir);
2762                 brelse(bh);
2763         } else
2764                 rc = -ENOENT;
2765
2766         cfs_up_write(&obj->oo_ext_idx_sem);
2767         LASSERT(osd_invariant(obj));
2768         RETURN(rc);
2769 }
2770
2771 /**
2772  *      Lookup index for \a key and copy record to \a rec.
2773  *
2774  *      \param  dt      osd index object
2775  *      \param  key     key for index
2776  *      \param  rec     record reference
2777  *
2778  *      \retval  +ve  success : exact mach
2779  *      \retval  0    return record with key not greater than \a key
2780  *      \retval -ve   failure
2781  */
2782 static int osd_index_iam_lookup(const struct lu_env *env, struct dt_object *dt,
2783                                 struct dt_rec *rec, const struct dt_key *key,
2784                                 struct lustre_capa *capa)
2785 {
2786         struct osd_object     *obj = osd_dt_obj(dt);
2787         struct iam_path_descr *ipd;
2788         struct iam_container  *bag = &obj->oo_dir->od_container;
2789         struct osd_thread_info *oti = osd_oti_get(env);
2790         struct iam_iterator    *it = &oti->oti_idx_it;
2791         struct iam_rec *iam_rec;
2792         int rc;
2793         ENTRY;
2794
2795         LASSERT(osd_invariant(obj));
2796         LASSERT(dt_object_exists(dt));
2797         LASSERT(bag->ic_object == obj->oo_inode);
2798
2799         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
2800                 RETURN(-EACCES);
2801
2802         ipd = osd_idx_ipd_get(env, bag);
2803         if (IS_ERR(ipd))
2804                 RETURN(-ENOMEM);
2805
2806         /* got ipd now we can start iterator. */
2807         iam_it_init(it, bag, 0, ipd);
2808
2809         rc = iam_it_get(it, (struct iam_key *)key);
2810         if (rc >= 0) {
2811                 if (S_ISDIR(obj->oo_inode->i_mode))
2812                         iam_rec = (struct iam_rec *)oti->oti_ldp;
2813                 else
2814                         iam_rec = (struct iam_rec *) rec;
2815
2816                 iam_reccpy(&it->ii_path.ip_leaf, (struct iam_rec *)iam_rec);
2817                 if (S_ISDIR(obj->oo_inode->i_mode))
2818                         osd_fid_unpack((struct lu_fid *) rec,
2819                                        (struct osd_fid_pack *)iam_rec);
2820         }
2821         iam_it_put(it);
2822         iam_it_fini(it);
2823         osd_ipd_put(env, bag, ipd);
2824
2825         LINVRNT(osd_invariant(obj));
2826
2827         RETURN(rc);
2828 }
2829
2830 /**
2831  *      Inserts (key, value) pair in \a dt index object.
2832  *
2833  *      \param  dt      osd index object
2834  *      \param  key     key for index
2835  *      \param  rec     record reference
2836  *      \param  th      transaction handler
2837  *
2838  *      \retval  0  success
2839  *      \retval -ve failure
2840  */
2841 static int osd_index_iam_insert(const struct lu_env *env, struct dt_object *dt,
2842                                 const struct dt_rec *rec, const struct dt_key *key,
2843                                 struct thandle *th, struct lustre_capa *capa,
2844                                 int ignore_quota)
2845 {
2846         struct osd_object     *obj = osd_dt_obj(dt);
2847         struct iam_path_descr *ipd;
2848         struct osd_thandle    *oh;
2849         struct iam_container  *bag = &obj->oo_dir->od_container;
2850 #ifdef HAVE_QUOTA_SUPPORT
2851         cfs_cap_t              save = current->cap_effective;
2852 #endif
2853         struct osd_thread_info *oti = osd_oti_get(env);
2854         struct iam_rec *iam_rec = (struct iam_rec *)oti->oti_ldp;
2855         int rc;
2856
2857         ENTRY;
2858
2859         LINVRNT(osd_invariant(obj));
2860         LASSERT(dt_object_exists(dt));
2861         LASSERT(bag->ic_object == obj->oo_inode);
2862         LASSERT(th != NULL);
2863
2864         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
2865                 return -EACCES;
2866
2867         ipd = osd_idx_ipd_get(env, bag);
2868         if (unlikely(ipd == NULL))
2869                 RETURN(-ENOMEM);
2870
2871         oh = container_of0(th, struct osd_thandle, ot_super);
2872         LASSERT(oh->ot_handle != NULL);
2873         LASSERT(oh->ot_handle->h_transaction != NULL);
2874 #ifdef HAVE_QUOTA_SUPPORT
2875         if (ignore_quota)
2876                 current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
2877         else
2878                 current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
2879 #endif
2880         if (S_ISDIR(obj->oo_inode->i_mode))
2881                 osd_fid_pack((struct osd_fid_pack *)iam_rec, rec, &oti->oti_fid);
2882         else
2883                 iam_rec = (struct iam_rec *) rec;
2884         rc = iam_insert(oh->ot_handle, bag, (const struct iam_key *)key,
2885                         iam_rec, ipd);
2886 #ifdef HAVE_QUOTA_SUPPORT
2887         current->cap_effective = save;
2888 #endif
2889         osd_ipd_put(env, bag, ipd);
2890         LINVRNT(osd_invariant(obj));
2891         RETURN(rc);
2892 }
2893
2894 /**
2895  * Calls ldiskfs_add_entry() to add directory entry
2896  * into the directory. This is required for
2897  * interoperability mode (b11826)
2898  *
2899  * \retval   0, on success
2900  * \retval -ve, on error
2901  */
2902 static int __osd_ea_add_rec(struct osd_thread_info *info,
2903                             struct osd_object *pobj,
2904                             struct inode  *cinode,
2905                             const char *name,
2906                             const struct dt_rec *fid,
2907                             struct thandle *th)
2908 {
2909         struct ldiskfs_dentry_param *ldp;
2910         struct dentry      *child;
2911         struct osd_thandle *oth;
2912         int rc;
2913
2914         oth = container_of(th, struct osd_thandle, ot_super);
2915         LASSERT(oth->ot_handle != NULL);
2916         LASSERT(oth->ot_handle->h_transaction != NULL);
2917
2918         child = osd_child_dentry_get(info->oti_env, pobj, name, strlen(name));
2919
2920         if (fid_is_igif((struct lu_fid *)fid) ||
2921             fid_seq((struct lu_fid *)fid) >= FID_SEQ_NORMAL) {
2922                 ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
2923                 osd_get_ldiskfs_dirent_param(ldp, fid);
2924                 child->d_fsdata = (void*) ldp;
2925         } else
2926                 child->d_fsdata = NULL;
2927         rc = ldiskfs_add_entry(oth->ot_handle, child, cinode);
2928
2929         RETURN(rc);
2930 }
2931
2932 /**
2933  * Calls ldiskfs_add_dot_dotdot() to add dot and dotdot entries
2934  * into the directory.Also sets flags into osd object to
2935  * indicate dot and dotdot are created. This is required for
2936  * interoperability mode (b11826)
2937  *
2938  * \param dir   directory for dot and dotdot fixup.
2939  * \param obj   child object for linking
2940  *
2941  * \retval   0, on success
2942  * \retval -ve, on error
2943  */
2944 static int osd_add_dot_dotdot(struct osd_thread_info *info,
2945                               struct osd_object *dir,
2946                               struct inode  *parent_dir, const char *name,
2947                               const struct dt_rec *dot_fid,
2948                               const struct dt_rec *dot_dot_fid,
2949                               struct thandle *th)
2950 {
2951         struct inode            *inode  = dir->oo_inode;
2952         struct ldiskfs_dentry_param *dot_ldp;
2953         struct ldiskfs_dentry_param *dot_dot_ldp;
2954         struct osd_thandle      *oth;
2955         int result = 0;
2956
2957         oth = container_of(th, struct osd_thandle, ot_super);
2958         LASSERT(oth->ot_handle->h_transaction != NULL);
2959         LASSERT(S_ISDIR(dir->oo_inode->i_mode));
2960
2961         if (strcmp(name, dot) == 0) {
2962                 if (dir->oo_compat_dot_created) {
2963                         result = -EEXIST;
2964                 } else {
2965                         LASSERT(inode == parent_dir);
2966                         dir->oo_compat_dot_created = 1;
2967                         result = 0;
2968                 }
2969         } else if(strcmp(name, dotdot) == 0) {
2970                 dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp;
2971                 dot_dot_ldp = (struct ldiskfs_dentry_param *)info->oti_ldp2;
2972
2973                 if (!dir->oo_compat_dot_created)
2974                         return -EINVAL;
2975                 if (fid_seq((struct lu_fid *)dot_fid) >= FID_SEQ_NORMAL) {
2976                         osd_get_ldiskfs_dirent_param(dot_ldp, dot_fid);
2977                         osd_get_ldiskfs_dirent_param(dot_dot_ldp, dot_dot_fid);
2978                 } else {
2979                         dot_ldp = NULL;
2980                         dot_dot_ldp = NULL;
2981                 }
2982                 /* in case of rename, dotdot is already created */
2983                 if (dir->oo_compat_dotdot_created) {
2984                         return __osd_ea_add_rec(info, dir, parent_dir, name,
2985                                                 dot_dot_fid, th);
2986                 }
2987
2988                 result = ldiskfs_add_dot_dotdot(oth->ot_handle, parent_dir, inode,
2989                                                 dot_ldp, dot_dot_ldp);
2990                 if (result == 0)
2991                        dir->oo_compat_dotdot_created = 1;
2992         }
2993
2994         return result;
2995 }
2996
2997
2998 /**
2999  * It will call the appropriate osd_add* function and return the
3000  * value, return by respective functions.
3001  */
3002 static int osd_ea_add_rec(const struct lu_env *env,
3003                           struct osd_object *pobj,
3004                           struct inode *cinode,
3005                           const char *name,
3006                           const struct dt_rec *fid,
3007                           struct thandle *th)
3008 {
3009         struct osd_thread_info    *info   = osd_oti_get(env);
3010         int rc;
3011
3012         if (name[0] == '.' && (name[1] == '\0' || (name[1] == '.' &&
3013                                                    name[2] =='\0')))
3014                 rc = osd_add_dot_dotdot(info, pobj, cinode, name,
3015                      (struct dt_rec *)lu_object_fid(&pobj->oo_dt.do_lu),
3016                                         fid, th);
3017         else
3018                 rc = __osd_ea_add_rec(info, pobj, cinode, name, fid, th);
3019
3020         return rc;
3021 }
3022
3023 /**
3024  * Calls ->lookup() to find dentry. From dentry get inode and
3025  * read inode's ea to get fid. This is required for  interoperability
3026  * mode (b11826)
3027  *
3028  * \retval   0, on success
3029  * \retval -ve, on error
3030  */
3031 static int osd_ea_lookup_rec(const struct lu_env *env, struct osd_object *obj,
3032                              struct dt_rec *rec, const struct dt_key *key)
3033 {
3034         struct inode               *dir    = obj->oo_inode;
3035         struct dentry              *dentry;
3036         struct ldiskfs_dir_entry_2 *de;
3037         struct buffer_head         *bh;
3038         struct lu_fid              *fid = (struct lu_fid *) rec;
3039         int ino;
3040         int rc;
3041
3042         LASSERT(dir->i_op != NULL && dir->i_op->lookup != NULL);
3043
3044         dentry = osd_child_dentry_get(env, obj,
3045                                       (char *)key, strlen((char *)key));
3046
3047         cfs_down_read(&obj->oo_ext_idx_sem);
3048         bh = ll_ldiskfs_find_entry(dir, dentry, &de);
3049         if (bh) {
3050                 ino = le32_to_cpu(de->inode);
3051                 rc = osd_get_fid_from_dentry(de, rec);
3052
3053                 /* done with de, release bh */
3054                 brelse(bh);
3055                 if (rc != 0)
3056                         rc = osd_ea_fid_get(env, obj, ino, fid);
3057         } else
3058                 rc = -ENOENT;
3059
3060         cfs_up_read(&obj->oo_ext_idx_sem);
3061         RETURN (rc);
3062 }
3063
3064 /**
3065  * Find the osd object for given fid.
3066  *
3067  * \param fid need to find the osd object having this fid
3068  *
3069  * \retval osd_object on success
3070  * \retval        -ve on error
3071  */
3072 struct osd_object *osd_object_find(const struct lu_env *env,
3073                                    struct dt_object *dt,
3074                                    const struct lu_fid *fid)
3075 {
3076         struct lu_device         *ludev = dt->do_lu.lo_dev;
3077         struct osd_object        *child = NULL;
3078         struct lu_object         *luch;
3079         struct lu_object         *lo;
3080
3081         luch = lu_object_find(env, ludev, fid, NULL);
3082         if (!IS_ERR(luch)) {
3083                 if (lu_object_exists(luch)) {
3084                         lo = lu_object_locate(luch->lo_header, ludev->ld_type);
3085                         if (lo != NULL)
3086                                 child = osd_obj(lo);
3087                         else
3088                                 LU_OBJECT_DEBUG(D_ERROR, env, luch,
3089                                                 "lu_object can't be located"
3090                                                 ""DFID"\n", PFID(fid));
3091
3092                         if (child == NULL) {
3093                                 lu_object_put(env, luch);
3094                                 CERROR("Unable to get osd_object\n");
3095                                 child = ERR_PTR(-ENOENT);
3096                         }
3097                 } else {
3098                         LU_OBJECT_DEBUG(D_ERROR, env, luch,
3099                                         "lu_object does not exists "DFID"\n",
3100                                         PFID(fid));
3101                         child = ERR_PTR(-ENOENT);
3102                 }
3103         } else
3104                 child = (void *)luch;
3105
3106         return child;
3107 }
3108
3109 /**
3110  * Put the osd object once done with it.
3111  *
3112  * \param obj osd object that needs to be put
3113  */
3114 static inline void osd_object_put(const struct lu_env *env,
3115                                   struct osd_object *obj)
3116 {
3117         lu_object_put(env, &obj->oo_dt.do_lu);
3118 }
3119
3120 /**
3121  * Index add function for interoperability mode (b11826).
3122  * It will add the directory entry.This entry is needed to
3123  * maintain name->fid mapping.
3124  *
3125  * \param key it is key i.e. file entry to be inserted
3126  * \param rec it is value of given key i.e. fid
3127  *
3128  * \retval   0, on success
3129  * \retval -ve, on error
3130  */
3131 static int osd_index_ea_insert(const struct lu_env *env, struct dt_object *dt,
3132                                const struct dt_rec *rec,
3133                                const struct dt_key *key, struct thandle *th,
3134                                struct lustre_capa *capa, int ignore_quota)
3135 {
3136         struct osd_object        *obj   = osd_dt_obj(dt);
3137         struct lu_fid            *fid   = (struct lu_fid *) rec;
3138         const char               *name  = (const char *)key;
3139         struct osd_object        *child;
3140 #ifdef HAVE_QUOTA_SUPPORT
3141         cfs_cap_t                 save  = current->cap_effective;
3142 #endif
3143         int rc;
3144
3145         ENTRY;
3146
3147         LASSERT(osd_invariant(obj));
3148         LASSERT(dt_object_exists(dt));
3149         LASSERT(th != NULL);
3150
3151         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_INSERT))
3152                 RETURN(-EACCES);
3153
3154         child = osd_object_find(env, dt, fid);
3155         if (!IS_ERR(child)) {
3156                 struct inode *inode = obj->oo_inode;
3157                 struct osd_thread_info *oti = osd_oti_get(env);
3158                 struct timespec *ctime = &oti->oti_time;
3159                 struct timespec *mtime = &oti->oti_time2;
3160
3161                 *ctime = inode->i_ctime;
3162                 *mtime = inode->i_mtime;
3163 #ifdef HAVE_QUOTA_SUPPORT
3164                 if (ignore_quota)
3165                         current->cap_effective |= CFS_CAP_SYS_RESOURCE_MASK;
3166                 else
3167                         current->cap_effective &= ~CFS_CAP_SYS_RESOURCE_MASK;
3168 #endif
3169                 cfs_down_write(&obj->oo_ext_idx_sem);
3170                 rc = osd_ea_add_rec(env, obj, child->oo_inode, name, rec, th);
3171                 cfs_up_write(&obj->oo_ext_idx_sem);
3172 #ifdef HAVE_QUOTA_SUPPORT
3173                 current->cap_effective = save;
3174 #endif
3175                 osd_object_put(env, child);
3176                 /* xtime should not be updated with server-side time. */
3177                 cfs_spin_lock(&obj->oo_guard);
3178                 inode->i_ctime = *ctime;
3179                 inode->i_mtime = *mtime;
3180                 cfs_spin_unlock(&obj->oo_guard);
3181                 mark_inode_dirty(inode);
3182         } else {
3183                 rc = PTR_ERR(child);
3184         }
3185
3186         LASSERT(osd_invariant(obj));
3187         RETURN(rc);
3188 }
3189
3190 /**
3191  *  Initialize osd Iterator for given osd index object.
3192  *
3193  *  \param  dt      osd index object
3194  */
3195
3196 static struct dt_it *osd_it_iam_init(const struct lu_env *env,
3197                                  struct dt_object *dt,
3198                                  struct lustre_capa *capa)
3199 {
3200         struct osd_it_iam         *it;
3201         struct osd_thread_info *oti = osd_oti_get(env);
3202         struct osd_object     *obj = osd_dt_obj(dt);
3203         struct lu_object      *lo  = &dt->do_lu;
3204         struct iam_path_descr *ipd;
3205         struct iam_container  *bag = &obj->oo_dir->od_container;
3206
3207         LASSERT(lu_object_exists(lo));
3208
3209         if (osd_object_auth(env, dt, capa, CAPA_OPC_BODY_READ))
3210                 return ERR_PTR(-EACCES);
3211
3212         it = &oti->oti_it;
3213         ipd = osd_it_ipd_get(env, bag);
3214         if (likely(ipd != NULL)) {
3215                 it->oi_obj = obj;
3216                 it->oi_ipd = ipd;
3217                 lu_object_get(lo);
3218                 iam_it_init(&it->oi_it, bag, IAM_IT_MOVE, ipd);
3219                 return (struct dt_it *)it;
3220         }
3221         return ERR_PTR(-ENOMEM);
3222 }
3223
3224 /**
3225  * free given Iterator.
3226  */
3227
3228 static void osd_it_iam_fini(const struct lu_env *env, struct dt_it *di)
3229 {
3230         struct osd_it_iam     *it = (struct osd_it_iam *)di;
3231         struct osd_object *obj = it->oi_obj;
3232
3233         iam_it_fini(&it->oi_it);
3234         osd_ipd_put(env, &obj->oo_dir->od_container, it->oi_ipd);
3235         lu_object_put(env, &obj->oo_dt.do_lu);
3236 }
3237
3238 /**
3239  *  Move Iterator to record specified by \a key
3240  *
3241  *  \param  di      osd iterator
3242  *  \param  key     key for index
3243  *
3244  *  \retval +ve  di points to record with least key not larger than key
3245  *  \retval  0   di points to exact matched key
3246  *  \retval -ve  failure
3247  */
3248
3249 static int osd_it_iam_get(const struct lu_env *env,
3250                       struct dt_it *di, const struct dt_key *key)
3251 {
3252         struct osd_it_iam *it = (struct osd_it_iam *)di;
3253
3254         return iam_it_get(&it->oi_it, (const struct iam_key *)key);
3255 }
3256
3257 /**
3258  *  Release Iterator
3259  *
3260  *  \param  di      osd iterator
3261  */
3262
3263 static void osd_it_iam_put(const struct lu_env *env, struct dt_it *di)
3264 {
3265         struct osd_it_iam *it = (struct osd_it_iam *)di;
3266
3267         iam_it_put(&it->oi_it);
3268 }
3269
3270 /**
3271  *  Move iterator by one record
3272  *
3273  *  \param  di      osd iterator
3274  *
3275  *  \retval +1   end of container reached
3276  *  \retval  0   success
3277  *  \retval -ve  failure
3278  */
3279
3280 static int osd_it_iam_next(const struct lu_env *env, struct dt_it *di)
3281 {
3282         struct osd_it_iam *it = (struct osd_it_iam *)di;
3283
3284         return iam_it_next(&it->oi_it);
3285 }
3286
3287 /**
3288  * Return pointer to the key under iterator.
3289  */
3290
3291 static struct dt_key *osd_it_iam_key(const struct lu_env *env,
3292                                  const struct dt_it *di)
3293 {
3294         struct osd_it_iam *it = (struct osd_it_iam *)di;
3295
3296         return (struct dt_key *)iam_it_key_get(&it->oi_it);
3297 }
3298
3299 /**
3300  * Return size of key under iterator (in bytes)
3301  */
3302
3303 static int osd_it_iam_key_size(const struct lu_env *env, const struct dt_it *di)
3304 {
3305         struct osd_it_iam *it = (struct osd_it_iam *)di;
3306
3307         return iam_it_key_size(&it->oi_it);
3308 }
3309
3310 static inline void osd_it_append_attrs(struct lu_dirent*ent,
3311                                        __u32 attr,
3312                                        int len,
3313                                        __u16 type)
3314 {
3315         struct luda_type        *lt;
3316         const unsigned           align = sizeof(struct luda_type) - 1;
3317
3318         /* check if file type is required */
3319         if (attr & LUDA_TYPE) {
3320                         len = (len + align) & ~align;
3321
3322                         lt = (void *) ent->lde_name + len;
3323                         lt->lt_type = cpu_to_le16(CFS_DTTOIF(type));
3324                         ent->lde_attrs |= LUDA_TYPE;
3325         }
3326
3327         ent->lde_attrs = cpu_to_le32(ent->lde_attrs);
3328 }
3329
3330 /**
3331  * build lu direct from backend fs dirent.
3332  */
3333
3334 static inline void osd_it_pack_dirent(struct lu_dirent *ent,
3335                                       struct lu_fid *fid,
3336                                       __u64 offset,
3337                                       char *name,
3338                                       __u16 namelen,
3339                                       __u16 type,
3340                                       __u32 attr)
3341 {
3342         fid_cpu_to_le(&ent->lde_fid, fid);
3343         ent->lde_attrs = LUDA_FID;
3344
3345         ent->lde_hash = cpu_to_le64(offset);
3346         ent->lde_reclen = cpu_to_le16(lu_dirent_calc_size(namelen, attr));
3347
3348         strncpy(ent->lde_name, name, namelen);
3349         ent->lde_namelen = cpu_to_le16(namelen);
3350
3351         /* append lustre attributes */
3352         osd_it_append_attrs(ent, attr, namelen, type);
3353 }
3354
3355 /**
3356  * Return pointer to the record under iterator.
3357  */
3358 static int osd_it_iam_rec(const struct lu_env *env,
3359                           const struct dt_it *di,
3360                           struct lu_dirent *lde,
3361                           __u32 attr)
3362 {
3363         struct osd_it_iam *it        = (struct osd_it_iam *)di;
3364         struct osd_thread_info *info = osd_oti_get(env);
3365         struct lu_fid     *fid       = &info->oti_fid;
3366         const struct osd_fid_pack *rec;
3367         char *name;
3368         int namelen;
3369         __u64 hash;
3370         int rc;
3371
3372         name = (char *)iam_it_key_get(&it->oi_it);
3373         if (IS_ERR(name))
3374                 RETURN(PTR_ERR(name));
3375
3376         namelen = iam_it_key_size(&it->oi_it);
3377
3378         rec = (const struct osd_fid_pack *) iam_it_rec_get(&it->oi_it);
3379         if (IS_ERR(rec))
3380                 RETURN(PTR_ERR(rec));
3381
3382         rc = osd_fid_unpack(fid, rec);
3383         if (rc)
3384                 RETURN(rc);
3385
3386         hash = iam_it_store(&it->oi_it);
3387
3388         /* IAM does not store object type in IAM index (dir) */
3389         osd_it_pack_dirent(lde, fid, hash, name, namelen,
3390                            0, LUDA_FID);
3391
3392         return 0;
3393 }
3394
3395 /**
3396  * Returns cookie for current Iterator position.
3397  */
3398 static __u64 osd_it_iam_store(const struct lu_env *env, const struct dt_it *di)
3399 {
3400         struct osd_it_iam *it = (struct osd_it_iam *)di;
3401
3402         return iam_it_store(&it->oi_it);
3403 }
3404
3405 /**
3406  * Restore iterator from cookie.
3407  *
3408  * \param  di      osd iterator
3409  * \param  hash    Iterator location cookie
3410  *
3411  * \retval +ve  di points to record with least key not larger than key.
3412  * \retval  0   di points to exact matched key
3413  * \retval -ve  failure
3414  */
3415
3416 static int osd_it_iam_load(const struct lu_env *env,
3417                        const struct dt_it *di, __u64 hash)
3418 {
3419         struct osd_it_iam *it = (struct osd_it_iam *)di;
3420
3421         return iam_it_load(&it->oi_it, hash);
3422 }
3423
3424 static const struct dt_index_operations osd_index_iam_ops = {
3425         .dio_lookup = osd_index_iam_lookup,
3426         .dio_insert = osd_index_iam_insert,
3427         .dio_delete = osd_index_iam_delete,
3428         .dio_it     = {
3429                 .init     = osd_it_iam_init,
3430                 .fini     = osd_it_iam_fini,
3431                 .get      = osd_it_iam_get,
3432                 .put      = osd_it_iam_put,
3433                 .next     = osd_it_iam_next,
3434                 .key      = osd_it_iam_key,
3435                 .key_size = osd_it_iam_key_size,
3436                 .rec      = osd_it_iam_rec,
3437                 .store    = osd_it_iam_store,
3438                 .load     = osd_it_iam_load
3439         }
3440 };
3441
3442 /**
3443  * Creates or initializes iterator context.
3444  *
3445  * \retval struct osd_it_ea, iterator structure on success
3446  *
3447  */
3448 static struct dt_it *osd_it_ea_init(const struct lu_env *env,
3449                                     struct dt_object *dt,
3450                                     struct lustre_capa *capa)
3451 {
3452         struct osd_object       *obj  = osd_dt_obj(dt);
3453         struct osd_thread_info  *info = osd_oti_get(env);
3454         struct osd_it_ea        *it   = &info->oti_it_ea;
3455         struct lu_object        *lo   = &dt->do_lu;
3456         struct dentry           *obj_dentry = &info->oti_it_dentry;
3457         ENTRY;
3458         LASSERT(lu_object_exists(lo));
3459
3460         obj_dentry->d_inode = obj->oo_inode;
3461         obj_dentry->d_sb = osd_sb(osd_obj2dev(obj));
3462         obj_dentry->d_name.hash = 0;
3463
3464         it->oie_rd_dirent       = 0;
3465         it->oie_it_dirent       = 0;
3466         it->oie_dirent          = NULL;
3467         it->oie_buf             = info->oti_it_ea_buf;
3468         it->oie_obj             = obj;
3469         it->oie_file.f_pos      = 0;
3470         it->oie_file.f_dentry   = obj_dentry;
3471         it->oie_file.f_mapping    = obj->oo_inode->i_mapping;
3472         it->oie_file.f_op         = obj->oo_inode->i_fop;
3473         it->oie_file.private_data = NULL;
3474         lu_object_get(lo);
3475         RETURN((struct dt_it *) it);
3476 }
3477
3478 /**
3479  * Destroy or finishes iterator context.
3480  *
3481  * \param di iterator structure to be destroyed
3482  */
3483 static void osd_it_ea_fini(const struct lu_env *env, struct dt_it *di)
3484 {
3485         struct osd_it_ea     *it   = (struct osd_it_ea *)di;
3486         struct osd_object    *obj  = it->oie_obj;
3487         struct inode       *inode  = obj->oo_inode;
3488
3489         ENTRY;
3490         it->oie_file.f_op->release(inode, &it->oie_file);
3491         lu_object_put(env, &obj->oo_dt.do_lu);
3492         EXIT;
3493 }
3494
3495 /**
3496  * It position the iterator at given key, so that next lookup continues from
3497  * that key Or it is similar to dio_it->load() but based on a key,
3498  * rather than file position.
3499  *
3500  * As a special convention, osd_it_ea_get(env, di, "") has to rewind iterator
3501  * to the beginning.
3502  *
3503  * TODO: Presently return +1 considering it is only used by mdd_dir_is_empty().
3504  */
3505 static int osd_it_ea_get(const struct lu_env *env,
3506                          struct dt_it *di, const struct dt_key *key)
3507 {
3508         struct osd_it_ea     *it   = (struct osd_it_ea *)di;
3509
3510         ENTRY;
3511         LASSERT(((const char *)key)[0] == '\0');
3512         it->oie_file.f_pos      = 0;
3513         it->oie_rd_dirent       = 0;
3514         it->oie_it_dirent       = 0;
3515         it->oie_dirent          = NULL;
3516
3517         RETURN(+1);
3518 }
3519
3520 /**
3521  * Does nothing
3522  */
3523 static void osd_it_ea_put(const struct lu_env *env, struct dt_it *di)
3524 {
3525 }
3526
3527 /**
3528  * It is called internally by ->readdir(). It fills the
3529  * iterator's in-memory data structure with required
3530  * information i.e. name, namelen, rec_size etc.
3531  *
3532  * \param buf in which information to be filled in.
3533  * \param name name of the file in given dir
3534  *
3535  * \retval 0 on success
3536  * \retval 1 on buffer full
3537  */
3538 static int osd_ldiskfs_filldir(char *buf, const char *name, int namelen,
3539                                loff_t offset, __u64 ino,
3540                                unsigned d_type)
3541 {
3542         struct osd_it_ea        *it   = (struct osd_it_ea *)buf;
3543         struct osd_it_ea_dirent *ent  = it->oie_dirent;
3544         struct lu_fid           *fid  = &ent->oied_fid;
3545         struct osd_fid_pack     *rec;
3546         ENTRY;
3547
3548         /* this should never happen */
3549         if (unlikely(namelen == 0 || namelen > LDISKFS_NAME_LEN)) {
3550                 CERROR("ldiskfs return invalid namelen %d\n", namelen);
3551                 RETURN(-EIO);
3552         }
3553
3554         if ((void *) ent - it->oie_buf + sizeof(*ent) + namelen >
3555             OSD_IT_EA_BUFSIZE)
3556                 RETURN(1);
3557
3558         if (d_type & LDISKFS_DIRENT_LUFID) {
3559                 rec = (struct osd_fid_pack*) (name + namelen + 1);
3560
3561                 if (osd_fid_unpack(fid, rec) != 0)
3562                         fid_zero(fid);
3563
3564                 d_type &= ~LDISKFS_DIRENT_LUFID;
3565         } else {
3566                 fid_zero(fid);
3567         }
3568
3569         ent->oied_ino     = ino;
3570         ent->oied_off     = offset;
3571         ent->oied_namelen = namelen;
3572         ent->oied_type    = d_type;
3573
3574         memcpy(ent->oied_name, name, namelen);
3575
3576         it->oie_rd_dirent++;
3577         it->oie_dirent = (void *) ent + cfs_size_round(sizeof(*ent) + namelen);
3578         RETURN(0);
3579 }
3580
3581 /**
3582  * Calls ->readdir() to load a directory entry at a time
3583  * and stored it in iterator's in-memory data structure.
3584  *
3585  * \param di iterator's in memory structure
3586  *
3587  * \retval   0 on success
3588  * \retval -ve on error
3589  */
3590 static int osd_ldiskfs_it_fill(const struct dt_it *di)
3591 {
3592         struct osd_it_ea   *it    = (struct osd_it_ea *)di;
3593         struct osd_object  *obj   = it->oie_obj;
3594         struct inode       *inode = obj->oo_inode;
3595         int                result = 0;
3596
3597         ENTRY;
3598         it->oie_dirent = it->oie_buf;
3599         it->oie_rd_dirent = 0;
3600
3601         cfs_down_read(&obj->oo_ext_idx_sem);
3602         result = inode->i_fop->readdir(&it->oie_file, it,
3603                                        (filldir_t) osd_ldiskfs_filldir);
3604
3605         cfs_up_read(&obj->oo_ext_idx_sem);
3606
3607         if (it->oie_rd_dirent == 0) {
3608                 result = -EIO;
3609         } else {
3610                 it->oie_dirent = it->oie_buf;
3611                 it->oie_it_dirent = 1;
3612         }
3613
3614         RETURN(result);
3615 }
3616
3617 /**
3618  * It calls osd_ldiskfs_it_fill() which will use ->readdir()
3619  * to load a directory entry at a time and stored it in
3620  * iterator's in-memory data structure.
3621  *
3622  * \param di iterator's in memory structure
3623  *
3624  * \retval +ve iterator reached to end
3625  * \retval   0 iterator not reached to end
3626  * \retval -ve on error
3627  */
3628 static int osd_it_ea_next(const struct lu_env *env, struct dt_it *di)
3629 {
3630         struct osd_it_ea *it = (struct osd_it_ea *)di;
3631         int rc;
3632
3633         ENTRY;
3634
3635         if (it->oie_it_dirent < it->oie_rd_dirent) {
3636                 it->oie_dirent =
3637                         (void *) it->oie_dirent +
3638                         cfs_size_round(sizeof(struct osd_it_ea_dirent) +
3639                                        it->oie_dirent->oied_namelen);
3640                 it->oie_it_dirent++;
3641                 RETURN(0);
3642         } else {
3643                 if (it->oie_file.f_pos == LDISKFS_HTREE_EOF)
3644                         rc = +1;
3645                 else
3646                         rc = osd_ldiskfs_it_fill(di);
3647         }
3648
3649         RETURN(rc);
3650 }
3651
3652 /**
3653  * Returns the key at current position from iterator's in memory structure.
3654  *
3655  * \param di iterator's in memory structure
3656  *
3657  * \retval key i.e. struct dt_key on success
3658  */
3659 static struct dt_key *osd_it_ea_key(const struct lu_env *env,
3660                                     const struct dt_it *di)
3661 {
3662         struct osd_it_ea *it = (struct osd_it_ea *)di;
3663         ENTRY;
3664         RETURN((struct dt_key *)it->oie_dirent->oied_name);
3665 }
3666
3667 /**
3668  * Returns the key's size at current position from iterator's in memory structure.
3669  *
3670  * \param di iterator's in memory structure
3671  *
3672  * \retval key_size i.e. struct dt_key on success
3673  */
3674 static int osd_it_ea_key_size(const struct lu_env *env, const struct dt_it *di)
3675 {
3676         struct osd_it_ea *it = (struct osd_it_ea *)di;
3677         ENTRY;
3678         RETURN(it->oie_dirent->oied_namelen);
3679 }
3680
3681
3682 /**
3683  * Returns the value (i.e. fid/igif) at current position from iterator's
3684  * in memory structure.
3685  *
3686  * \param di struct osd_it_ea, iterator's in memory structure
3687  * \param attr attr requested for dirent.
3688  * \param lde lustre dirent
3689  *
3690  * \retval   0 no error and \param lde has correct lustre dirent.
3691  * \retval -ve on error
3692  */
3693 static inline int osd_it_ea_rec(const struct lu_env *env,
3694                                 const struct dt_it *di,
3695                                 struct lu_dirent *lde,
3696                                 __u32 attr)
3697 {
3698         struct osd_it_ea        *it     = (struct osd_it_ea *)di;
3699         struct osd_object       *obj    = it->oie_obj;
3700         struct lu_fid           *fid    = &it->oie_dirent->oied_fid;
3701         int    rc = 0;
3702
3703         ENTRY;
3704
3705         if (!fid_is_sane(fid))
3706                 rc = osd_ea_fid_get(env, obj, it->oie_dirent->oied_ino, fid);
3707
3708         if (rc == 0)
3709                 osd_it_pack_dirent(lde, fid, it->oie_dirent->oied_off,
3710                                    it->oie_dirent->oied_name,
3711                                    it->oie_dirent->oied_namelen,
3712                                    it->oie_dirent->oied_type,
3713                                    attr);
3714         RETURN(rc);
3715 }
3716
3717 /**
3718  * Returns a cookie for current position of the iterator head, so that
3719  * user can use this cookie to load/start the iterator next time.
3720  *
3721  * \param di iterator's in memory structure
3722  *
3723  * \retval cookie for current position, on success
3724  */
3725 static __u64 osd_it_ea_store(const struct lu_env *env, const struct dt_it *di)
3726 {
3727         struct osd_it_ea *it = (struct osd_it_ea *)di;
3728         ENTRY;
3729         RETURN(it->oie_dirent->oied_off);
3730 }
3731
3732 /**
3733  * It calls osd_ldiskfs_it_fill() which will use ->readdir()
3734  * to load a directory entry at a time and stored it i inn,
3735  * in iterator's in-memory data structure.
3736  *
3737  * \param di struct osd_it_ea, iterator's in memory structure
3738  *
3739  * \retval +ve on success
3740  * \retval -ve on error
3741  */
3742 static int osd_it_ea_load(const struct lu_env *env,
3743                           const struct dt_it *di, __u64 hash)
3744 {
3745         struct osd_it_ea *it = (struct osd_it_ea *)di;
3746         int rc;
3747
3748         ENTRY;
3749         it->oie_file.f_pos = hash;
3750
3751         rc =  osd_ldiskfs_it_fill(di);
3752         if (rc == 0)
3753                 rc = +1;
3754
3755         RETURN(rc);
3756 }
3757
3758 /**
3759  * Index lookup function for interoperability mode (b11826).
3760  *
3761  * \param key,  key i.e. file name to be searched
3762  *
3763  * \retval +ve, on success
3764  * \retval -ve, on error
3765  */
3766 static int osd_index_ea_lookup(const struct lu_env *env, struct dt_object *dt,
3767                                struct dt_rec *rec, const struct dt_key *key,
3768                                struct lustre_capa *capa)
3769 {
3770         struct osd_object *obj = osd_dt_obj(dt);
3771         int rc = 0;
3772
3773         ENTRY;
3774
3775         LASSERT(S_ISDIR(obj->oo_inode->i_mode));
3776         LINVRNT(osd_invariant(obj));
3777
3778         if (osd_object_auth(env, dt, capa, CAPA_OPC_INDEX_LOOKUP))
3779                 return -EACCES;
3780
3781         rc = osd_ea_lookup_rec(env, obj, rec, key);
3782
3783         if (rc == 0)
3784                 rc = +1;
3785         RETURN(rc);
3786 }
3787
3788 /**
3789  * Index and Iterator operations for interoperability
3790  * mode (i.e. to run 2.0 mds on 1.8 disk) (b11826)
3791  */
3792 static const struct dt_index_operations osd_index_ea_ops = {
3793         .dio_lookup = osd_index_ea_lookup,
3794         .dio_insert = osd_index_ea_insert,
3795         .dio_delete = osd_index_ea_delete,
3796         .dio_it     = {
3797                 .init     = osd_it_ea_init,
3798                 .fini     = osd_it_ea_fini,
3799                 .get      = osd_it_ea_get,
3800                 .put      = osd_it_ea_put,
3801                 .next     = osd_it_ea_next,
3802                 .key      = osd_it_ea_key,
3803                 .key_size = osd_it_ea_key_size,
3804                 .rec      = osd_it_ea_rec,
3805                 .store    = osd_it_ea_store,
3806                 .load     = osd_it_ea_load
3807         }
3808 };
3809
3810 static void *osd_key_init(const struct lu_context *ctx,
3811                           struct lu_context_key *key)
3812 {
3813         struct osd_thread_info *info;
3814
3815         OBD_ALLOC_PTR(info);
3816         if (info != NULL) {
3817                 OBD_ALLOC(info->oti_it_ea_buf, OSD_IT_EA_BUFSIZE);
3818                 if (info->oti_it_ea_buf != NULL) {
3819                         info->oti_env = container_of(ctx, struct lu_env,
3820                                                      le_ctx);
3821                 } else {
3822                         OBD_FREE_PTR(info);
3823                         info = ERR_PTR(-ENOMEM);
3824                 }
3825         } else {
3826                 info = ERR_PTR(-ENOMEM);
3827         }
3828         return info;
3829 }
3830
3831 static void osd_key_fini(const struct lu_context *ctx,
3832                          struct lu_context_key *key, void* data)
3833 {
3834         struct osd_thread_info *info = data;
3835
3836         OBD_FREE(info->oti_it_ea_buf, OSD_IT_EA_BUFSIZE);
3837         OBD_FREE_PTR(info);
3838 }
3839
3840 static void osd_key_exit(const struct lu_context *ctx,
3841                          struct lu_context_key *key, void *data)
3842 {
3843         struct osd_thread_info *info = data;
3844
3845         LASSERT(info->oti_r_locks == 0);
3846         LASSERT(info->oti_w_locks == 0);
3847         LASSERT(info->oti_txns    == 0);
3848 }
3849
3850 /* type constructor/destructor: osd_type_init, osd_type_fini */
3851 LU_TYPE_INIT_FINI(osd, &osd_key);
3852
3853 static struct lu_context_key osd_key = {
3854         .lct_tags = LCT_DT_THREAD | LCT_MD_THREAD,
3855         .lct_init = osd_key_init,
3856         .lct_fini = osd_key_fini,
3857         .lct_exit = osd_key_exit
3858 };
3859
3860
3861 static int osd_device_init(const struct lu_env *env, struct lu_device *d,
3862                            const char *name, struct lu_device *next)
3863 {
3864         int rc;
3865         struct lu_context *ctx;
3866
3867         /* context for commit hooks */
3868         ctx = &osd_dev(d)->od_env_for_commit.le_ctx;
3869         rc = lu_context_init(ctx, LCT_MD_THREAD|LCT_REMEMBER|LCT_NOREF);
3870         if (rc == 0) {
3871                 rc = osd_procfs_init(osd_dev(d), name);
3872                 ctx->lc_cookie = 0x3;
3873         }
3874         return rc;
3875 }
3876
3877 static int osd_shutdown(const struct lu_env *env, struct osd_device *o)
3878 {
3879         struct osd_thread_info *info = osd_oti_get(env);
3880         ENTRY;
3881         if (o->od_obj_area != NULL) {
3882                 lu_object_put(env, &o->od_obj_area->do_lu);
3883                 o->od_obj_area = NULL;
3884         }
3885         osd_oi_fini(info, &o->od_oi);
3886
3887         RETURN(0);
3888 }
3889
3890 static int osd_mount(const struct lu_env *env,
3891                      struct osd_device *o, struct lustre_cfg *cfg)
3892 {
3893         struct lustre_mount_info *lmi;
3894         const char               *dev  = lustre_cfg_string(cfg, 0);
3895         struct lustre_disk_data  *ldd;
3896         struct lustre_sb_info    *lsi;
3897
3898         ENTRY;
3899         if (o->od_mount != NULL) {
3900                 CERROR("Already mounted (%s)\n", dev);
3901                 RETURN(-EEXIST);
3902         }
3903
3904         /* get mount */
3905         lmi = server_get_mount(dev);
3906         if (lmi == NULL) {
3907                 CERROR("Cannot get mount info for %s!\n", dev);
3908                 RETURN(-EFAULT);
3909         }
3910
3911         LASSERT(lmi != NULL);
3912         /* save lustre_mount_info in dt_device */
3913         o->od_mount = lmi;
3914
3915         lsi = s2lsi(lmi->lmi_sb);
3916         ldd = lsi->lsi_ldd;
3917
3918         if (ldd->ldd_flags & LDD_F_IAM_DIR) {
3919                 o->od_iop_mode = 0;
3920                 LCONSOLE_WARN("OSD: IAM mode enabled\n");
3921         } else
3922                 o->od_iop_mode = 1;
3923
3924         o->od_obj_area = NULL;
3925         RETURN(0);
3926 }
3927
3928 static struct lu_device *osd_device_fini(const struct lu_env *env,
3929                                          struct lu_device *d)
3930 {
3931         int rc;
3932         ENTRY;
3933
3934         shrink_dcache_sb(osd_sb(osd_dev(d)));
3935         osd_sync(env, lu2dt_dev(d));
3936
3937         rc = osd_procfs_fini(osd_dev(d));
3938         if (rc) {
3939                 CERROR("proc fini error %d \n", rc);
3940                 RETURN (ERR_PTR(rc));
3941         }
3942
3943         if (osd_dev(d)->od_mount)
3944                 server_put_mount(osd_dev(d)->od_mount->lmi_name,
3945                                  osd_dev(d)->od_mount->lmi_mnt);
3946         osd_dev(d)->od_mount = NULL;
3947
3948         lu_context_fini(&osd_dev(d)->od_env_for_commit.le_ctx);
3949         RETURN(NULL);
3950 }
3951
3952 static struct lu_device *osd_device_alloc(const struct lu_env *env,
3953                                           struct lu_device_type *t,
3954                                           struct lustre_cfg *cfg)
3955 {
3956         struct lu_device  *l;
3957         struct osd_device *o;
3958
3959         OBD_ALLOC_PTR(o);
3960         if (o != NULL) {
3961                 int result;
3962
3963                 result = dt_device_init(&o->od_dt_dev, t);
3964                 if (result == 0) {
3965                         l = osd2lu_dev(o);
3966                         l->ld_ops = &osd_lu_ops;
3967                         o->od_dt_dev.dd_ops = &osd_dt_ops;
3968                         cfs_spin_lock_init(&o->od_osfs_lock);
3969                         o->od_osfs_age = cfs_time_shift_64(-1000);
3970                         o->od_capa_hash = init_capa_hash();
3971                         if (o->od_capa_hash == NULL) {
3972                                 dt_device_fini(&o->od_dt_dev);
3973                                 l = ERR_PTR(-ENOMEM);
3974                         }
3975                 } else
3976                         l = ERR_PTR(result);
3977
3978                 if (IS_ERR(l))
3979                         OBD_FREE_PTR(o);
3980         } else
3981                 l = ERR_PTR(-ENOMEM);
3982         return l;
3983 }
3984
3985 static struct lu_device *osd_device_free(const struct lu_env *env,
3986                                          struct lu_device *d)
3987 {
3988         struct osd_device *o = osd_dev(d);
3989         ENTRY;
3990
3991         cleanup_capa_hash(o->od_capa_hash);
3992         dt_device_fini(&o->od_dt_dev);
3993         OBD_FREE_PTR(o);
3994         RETURN(NULL);
3995 }
3996
3997 static int osd_process_config(const struct lu_env *env,
3998                               struct lu_device *d, struct lustre_cfg *cfg)
3999 {
4000         struct osd_device *o = osd_dev(d);
4001         int err;
4002         ENTRY;
4003
4004         switch(cfg->lcfg_command) {
4005         case LCFG_SETUP:
4006                 err = osd_mount(env, o, cfg);
4007                 break;
4008         case LCFG_CLEANUP:
4009                 err = osd_shutdown(env, o);
4010                 break;
4011         default:
4012                 err = -ENOSYS;
4013         }
4014
4015         RETURN(err);
4016 }
4017
4018 static int osd_recovery_complete(const struct lu_env *env,
4019                                  struct lu_device *d)
4020 {
4021         RETURN(0);
4022 }
4023
4024 static int osd_prepare(const struct lu_env *env,
4025                        struct lu_device *pdev,
4026                        struct lu_device *dev)
4027 {
4028         struct osd_device *osd = osd_dev(dev);
4029         struct lustre_sb_info *lsi;
4030         struct lustre_disk_data *ldd;
4031         struct lustre_mount_info  *lmi;
4032         struct osd_thread_info *oti = osd_oti_get(env);
4033         struct dt_object *d;
4034         int result;
4035
4036         ENTRY;
4037         /* 1. initialize oi before any file create or file open */
4038         result = osd_oi_init(oti, &osd->od_oi,
4039                              &osd->od_dt_dev, lu2md_dev(pdev));
4040         if (result != 0)
4041                 RETURN(result);
4042
4043         lmi = osd->od_mount;
4044         lsi = s2lsi(lmi->lmi_sb);
4045         ldd = lsi->lsi_ldd;
4046
4047         /* 2. setup local objects */
4048         result = llo_local_objects_setup(env, lu2md_dev(pdev), lu2dt_dev(dev));
4049         if (result)
4050                 goto out;
4051
4052         /* 3. open remote object dir */
4053         d = dt_store_open(env, lu2dt_dev(dev), "",
4054                           remote_obj_dir, &oti->oti_fid);
4055         if (!IS_ERR(d)) {
4056                 osd->od_obj_area = d;
4057                 result = 0;
4058         } else {
4059                 result = PTR_ERR(d);
4060                 osd->od_obj_area = NULL;
4061         }
4062
4063 out:
4064         RETURN(result);
4065 }
4066
4067 static const struct lu_object_operations osd_lu_obj_ops = {
4068         .loo_object_init      = osd_object_init,
4069         .loo_object_delete    = osd_object_delete,
4070         .loo_object_release   = osd_object_release,
4071         .loo_object_free      = osd_object_free,
4072         .loo_object_print     = osd_object_print,
4073         .loo_object_invariant = osd_object_invariant
4074 };
4075
4076 static const struct lu_device_operations osd_lu_ops = {
4077         .ldo_object_alloc      = osd_object_alloc,
4078         .ldo_process_config    = osd_process_config,
4079         .ldo_recovery_complete = osd_recovery_complete,
4080         .ldo_prepare           = osd_prepare,
4081 };
4082
4083 static const struct lu_device_type_operations osd_device_type_ops = {
4084         .ldto_init = osd_type_init,
4085         .ldto_fini = osd_type_fini,
4086
4087         .ldto_start = osd_type_start,
4088         .ldto_stop  = osd_type_stop,
4089
4090         .ldto_device_alloc = osd_device_alloc,
4091         .ldto_device_free  = osd_device_free,
4092
4093         .ldto_device_init    = osd_device_init,
4094         .ldto_device_fini    = osd_device_fini
4095 };
4096
4097 static struct lu_device_type osd_device_type = {
4098         .ldt_tags     = LU_DEVICE_DT,
4099         .ldt_name     = LUSTRE_OSD_NAME,
4100         .ldt_ops      = &osd_device_type_ops,
4101         .ldt_ctx_tags = LCT_MD_THREAD|LCT_DT_THREAD
4102 };
4103
4104 /*
4105  * lprocfs legacy support.
4106  */
4107 static struct obd_ops osd_obd_device_ops = {
4108         .o_owner = THIS_MODULE
4109 };
4110
4111 static struct lu_local_obj_desc llod_osd_rem_obj_dir = {
4112         .llod_name      = remote_obj_dir,
4113         .llod_oid       = OSD_REM_OBJ_DIR_OID,
4114         .llod_is_index  = 1,
4115         .llod_feat      = &dt_directory_features,
4116 };
4117
4118 static int __init osd_mod_init(void)
4119 {
4120         struct lprocfs_static_vars lvars;
4121
4122         osd_oi_mod_init();
4123         llo_local_obj_register(&llod_osd_rem_obj_dir);
4124         lprocfs_osd_init_vars(&lvars);
4125         return class_register_type(&osd_obd_device_ops, NULL, lvars.module_vars,
4126                                    LUSTRE_OSD_NAME, &osd_device_type);
4127 }
4128
4129 static void __exit osd_mod_exit(void)
4130 {
4131         llo_local_obj_unregister(&llod_osd_rem_obj_dir);
4132         class_unregister_type(LUSTRE_OSD_NAME);
4133 }
4134
4135 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
4136 MODULE_DESCRIPTION("Lustre Object Storage Device ("LUSTRE_OSD_NAME")");
4137 MODULE_LICENSE("GPL");
4138
4139 cfs_module(osd, "0.0.2", osd_mod_init, osd_mod_exit);