lustre/include/dt_object.h

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef __LUSTRE_DT_OBJECT_H
  38 #define __LUSTRE_DT_OBJECT_H
  39
  40 /** \defgroup dt dt
  41  * Sub-class of lu_object with methods common for "data" objects in OST stack.
  42  *
  43  * Data objects behave like regular files: you can read/write them, get and
  44  * set their attributes. Implementation of dt interface is supposed to
  45  * implement some form of garbage collection, normally reference counting
  46  * (nlink) based one.
  47  *
  48  * Examples: osd (lustre/osd) is an implementation of dt interface.
  49  * @{
  50  */
  51
  52 #include <obd_support.h>
  53 /*
  54  * super-class definitions.
  55  */
  56 #include <lu_object.h>
  57
  58 #include <libcfs/libcfs.h>
  59
  60 struct seq_file;
  61 struct proc_dir_entry;
  62 struct lustre_cfg;
  63
  64 struct thandle;
  65 struct dt_device;
  66 struct dt_object;
  67 struct dt_index_features;
  68 struct niobuf_local;
  69 struct niobuf_remote;
  70 struct ldlm_enqueue_info;
  71
  72 typedef enum {
  73         MNTOPT_USERXATTR        = 0x00000001,
  74         MNTOPT_ACL              = 0x00000002,
  75 } mntopt_t;
  76
  77 struct dt_device_param {
  78         unsigned           ddp_max_name_len;
  79         unsigned           ddp_max_nlink;
  80         unsigned           ddp_block_shift;
  81         mntopt_t           ddp_mntopts;
  82         unsigned           ddp_max_ea_size;
  83         int                ddp_mount_type;
  84         unsigned long long ddp_maxbytes;
  85         /* percentage of available space to reserve for grant error margin */
  86         int                ddp_grant_reserved;
  87         /* per-inode space consumption */
  88         short              ddp_inodespace;
  89         /* per-fragment grant overhead to be used by client for grant
  90          * calculation */
  91         int                ddp_grant_frag;
  92 };
  93
  94 /**
  95  * Per-transaction commit callback function
  96  */
  97 struct dt_txn_commit_cb;
  98 typedef void (*dt_cb_t)(struct lu_env *env, struct thandle *th,
  99                         struct dt_txn_commit_cb *cb, int err);
 100 /**
 101  * Special per-transaction callback for cases when just commit callback
 102  * is needed and per-device callback are not convenient to use
 103  */
 104 #define TRANS_COMMIT_CB_MAGIC   0xa0a00a0a
 105 #define MAX_COMMIT_CB_STR_LEN   32
 106
 107 struct dt_txn_commit_cb {
 108         struct list_head        dcb_linkage;
 109         dt_cb_t                 dcb_func;
 110         __u32                   dcb_magic;
 111         char                    dcb_name[MAX_COMMIT_CB_STR_LEN];
 112 };
 113
 114 /**
 115  * Operations on dt device.
 116  */
 117 struct dt_device_operations {
 118         /**
 119          * Return device-wide statistics.
 120          */
 121         int   (*dt_statfs)(const struct lu_env *env,
 122                            struct dt_device *dev, struct obd_statfs *osfs);
 123         /**
 124          * Create transaction, described by \a param.
 125          */
 126         struct thandle *(*dt_trans_create)(const struct lu_env *env,
 127                                            struct dt_device *dev);
 128         /**
 129          * Start transaction, described by \a param.
 130          */
 131         int   (*dt_trans_start)(const struct lu_env *env,
 132                                 struct dt_device *dev, struct thandle *th);
 133         /**
 134          * Finish previously started transaction.
 135          */
 136         int   (*dt_trans_stop)(const struct lu_env *env, struct dt_device *dev,
 137                                struct thandle *th);
 138         /**
 139          * Add commit callback to the transaction.
 140          */
 141         int   (*dt_trans_cb_add)(struct thandle *th,
 142                                  struct dt_txn_commit_cb *dcb);
 143         /**
 144          * Return fid of root index object.
 145          */
 146         int   (*dt_root_get)(const struct lu_env *env,
 147                              struct dt_device *dev, struct lu_fid *f);
 148         /**
 149          * Return device configuration data.
 150          */
 151         void  (*dt_conf_get)(const struct lu_env *env,
 152                              const struct dt_device *dev,
 153                              struct dt_device_param *param);
 154         /**
 155          *  handling device state, mostly for tests
 156          */
 157         int   (*dt_sync)(const struct lu_env *env, struct dt_device *dev);
 158         int   (*dt_ro)(const struct lu_env *env, struct dt_device *dev);
 159         /**
 160           * Start a transaction commit asynchronously
 161           *
 162           * \param env environment
 163           * \param dev dt_device to start commit on
 164           *
 165           * \return 0 success, negative value if error
 166           */
 167          int   (*dt_commit_async)(const struct lu_env *env,
 168                                   struct dt_device *dev);
 169         /**
 170          * Initialize capability context.
 171          */
 172         int   (*dt_init_capa_ctxt)(const struct lu_env *env,
 173                                    struct dt_device *dev,
 174                                    int mode, unsigned long timeout,
 175                                    __u32 alg, struct lustre_capa_key *keys);
 176 };
 177
 178 struct dt_index_features {
 179         /** required feature flags from enum dt_index_flags */
 180         __u32 dif_flags;
 181         /** minimal required key size */
 182         size_t dif_keysize_min;
 183         /** maximal required key size, 0 if no limit */
 184         size_t dif_keysize_max;
 185         /** minimal required record size */
 186         size_t dif_recsize_min;
 187         /** maximal required record size, 0 if no limit */
 188         size_t dif_recsize_max;
 189         /** pointer size for record */
 190         size_t dif_ptrsize;
 191 };
 192
 193 enum dt_index_flags {
 194         /** index supports variable sized keys */
 195         DT_IND_VARKEY = 1 << 0,
 196         /** index supports variable sized records */
 197         DT_IND_VARREC = 1 << 1,
 198         /** index can be modified */
 199         DT_IND_UPDATE = 1 << 2,
 200         /** index supports records with non-unique (duplicate) keys */
 201         DT_IND_NONUNQ = 1 << 3,
 202         /**
 203          * index support fixed-size keys sorted with natural numerical way
 204          * and is able to return left-side value if no exact value found
 205          */
 206         DT_IND_RANGE = 1 << 4,
 207 };
 208
 209 /**
 210  * Features, required from index to support file system directories (mapping
 211  * names to fids).
 212  */
 213 extern const struct dt_index_features dt_directory_features;
 214 extern const struct dt_index_features dt_otable_features;
 215 extern const struct dt_index_features dt_lfsck_orphan_features;
 216 extern const struct dt_index_features dt_lfsck_features;
 217
 218 /* index features supported by the accounting objects */
 219 extern const struct dt_index_features dt_acct_features;
 220
 221 /* index features supported by the quota global indexes */
 222 extern const struct dt_index_features dt_quota_glb_features;
 223
 224 /* index features supported by the quota slave indexes */
 225 extern const struct dt_index_features dt_quota_slv_features;
 226
 227 /**
 228  * This is a general purpose dt allocation hint.
 229  * It now contains the parent object.
 230  * It can contain any allocation hint in the future.
 231  */
 232 struct dt_allocation_hint {
 233         struct dt_object        *dah_parent;
 234         const void              *dah_eadata;
 235         int                     dah_eadata_len;
 236         __u32                   dah_mode;
 237 };
 238
 239 /**
 240  * object type specifier.
 241  */
 242
 243 enum dt_format_type {
 244         DFT_REGULAR,
 245         DFT_DIR,
 246         /** for mknod */
 247         DFT_NODE,
 248         /** for special index */
 249         DFT_INDEX,
 250         /** for symbolic link */
 251         DFT_SYM,
 252 };
 253
 254 /**
 255  * object format specifier.
 256  */
 257 struct dt_object_format {
 258         /** type for dt object */
 259         enum dt_format_type dof_type;
 260         union {
 261                 struct dof_regular {
 262                         int striped;
 263                 } dof_reg;
 264                 struct dof_dir {
 265                 } dof_dir;
 266                 struct dof_node {
 267                 } dof_node;
 268                 /**
 269                  * special index need feature as parameter to create
 270                  * special idx
 271                  */
 272                 struct dof_index {
 273                         const struct dt_index_features *di_feat;
 274                 } dof_idx;
 275         } u;
 276 };
 277
 278 enum dt_format_type dt_mode_to_dft(__u32 mode);
 279
 280 typedef __u64 dt_obj_version_t;
 281
 282 union ldlm_policy_data;
 283
 284 /**
 285  * Per-dt-object operations.
 286  */
 287 struct dt_object_operations {
 288         void  (*do_read_lock)(const struct lu_env *env,
 289                               struct dt_object *dt, unsigned role);
 290         void  (*do_write_lock)(const struct lu_env *env,
 291                                struct dt_object *dt, unsigned role);
 292         void  (*do_read_unlock)(const struct lu_env *env,
 293                                 struct dt_object *dt);
 294         void  (*do_write_unlock)(const struct lu_env *env,
 295                                  struct dt_object *dt);
 296         int  (*do_write_locked)(const struct lu_env *env,
 297                                 struct dt_object *dt);
 298         /**
 299          * Note: following ->do_{x,}attr_{set,get}() operations are very
 300          * similar to ->moo_{x,}attr_{set,get}() operations in struct
 301          * md_object_operations (see md_object.h). These operations are not in
 302          * lu_object_operations, because ->do_{x,}attr_set() versions take
 303          * transaction handle as an argument (this transaction is started by
 304          * caller). We might factor ->do_{x,}attr_get() into
 305          * lu_object_operations, but that would break existing symmetry.
 306          */
 307
 308         int   (*do_declare_attr_get)(const struct lu_env *env,
 309                                      struct dt_object *dt,
 310                                      struct lustre_capa *capa);
 311         /**
 312          * Return standard attributes.
 313          *
 314          * precondition: lu_object_exists(&dt->do_lu);
 315          */
 316         int   (*do_attr_get)(const struct lu_env *env,
 317                              struct dt_object *dt, struct lu_attr *attr,
 318                              struct lustre_capa *capa);
 319         /**
 320          * Set standard attributes.
 321          *
 322          * precondition: dt_object_exists(dt);
 323          */
 324         int   (*do_declare_attr_set)(const struct lu_env *env,
 325                                      struct dt_object *dt,
 326                                      const struct lu_attr *attr,
 327                                      struct thandle *handle);
 328         int   (*do_attr_set)(const struct lu_env *env,
 329                              struct dt_object *dt,
 330                              const struct lu_attr *attr,
 331                              struct thandle *handle,
 332                              struct lustre_capa *capa);
 333
 334         int   (*do_declare_xattr_get)(const struct lu_env *env,
 335                                       struct dt_object *dt,
 336                                       struct lu_buf *buf,
 337                                       const char *name,
 338                                       struct lustre_capa *capa);
 339
 340         /**
 341          * Return a value of an extended attribute.
 342          *
 343          * precondition: dt_object_exists(dt);
 344          */
 345         int   (*do_xattr_get)(const struct lu_env *env, struct dt_object *dt,
 346                               struct lu_buf *buf, const char *name,
 347                               struct lustre_capa *capa);
 348         /**
 349          * Set value of an extended attribute.
 350          *
 351          * \a fl - flags from enum lu_xattr_flags
 352          *
 353          * precondition: dt_object_exists(dt);
 354          */
 355         int   (*do_declare_xattr_set)(const struct lu_env *env,
 356                                       struct dt_object *dt,
 357                                       const struct lu_buf *buf,
 358                                       const char *name, int fl,
 359                                       struct thandle *handle);
 360         int   (*do_xattr_set)(const struct lu_env *env,
 361                               struct dt_object *dt, const struct lu_buf *buf,
 362                               const char *name, int fl, struct thandle *handle,
 363                               struct lustre_capa *capa);
 364         /**
 365          * Delete existing extended attribute.
 366          *
 367          * precondition: dt_object_exists(dt);
 368          */
 369         int   (*do_declare_xattr_del)(const struct lu_env *env,
 370                                       struct dt_object *dt,
 371                                       const char *name, struct thandle *handle);
 372         int   (*do_xattr_del)(const struct lu_env *env,
 373                               struct dt_object *dt,
 374                               const char *name, struct thandle *handle,
 375                               struct lustre_capa *capa);
 376         /**
 377          * Place list of existing extended attributes into \a buf (which has
 378          * length len).
 379          *
 380          * precondition: dt_object_exists(dt);
 381          */
 382         int   (*do_xattr_list)(const struct lu_env *env,
 383                                struct dt_object *dt, struct lu_buf *buf,
 384                                struct lustre_capa *capa);
 385         /**
 386          * Init allocation hint using parent object and child mode.
 387          * (1) The \a parent might be NULL if this is a partial creation for
 388          *     remote object.
 389          * (2) The type of child is in \a child_mode.
 390          * (3) The result hint is stored in \a ah;
 391          */
 392         void  (*do_ah_init)(const struct lu_env *env,
 393                             struct dt_allocation_hint *ah,
 394                             struct dt_object *parent,
 395                             struct dt_object *child,
 396                             umode_t child_mode);
 397         /**
 398          * Create new object on this device.
 399          *
 400          * precondition: !dt_object_exists(dt);
 401          * postcondition: ergo(result == 0, dt_object_exists(dt));
 402          */
 403         int   (*do_declare_create)(const struct lu_env *env,
 404                                    struct dt_object *dt,
 405                                    struct lu_attr *attr,
 406                                    struct dt_allocation_hint *hint,
 407                                    struct dt_object_format *dof,
 408                                    struct thandle *th);
 409         int   (*do_create)(const struct lu_env *env, struct dt_object *dt,
 410                            struct lu_attr *attr,
 411                            struct dt_allocation_hint *hint,
 412                            struct dt_object_format *dof,
 413                            struct thandle *th);
 414
 415         /**
 416           Destroy object on this device
 417          * precondition: !dt_object_exists(dt);
 418          * postcondition: ergo(result == 0, dt_object_exists(dt));
 419          */
 420         int   (*do_declare_destroy)(const struct lu_env *env,
 421                                     struct dt_object *dt,
 422                                     struct thandle *th);
 423         int   (*do_destroy)(const struct lu_env *env, struct dt_object *dt,
 424                             struct thandle *th);
 425
 426         /**
 427          * Announce that this object is going to be used as an index. This
 428          * operation check that object supports indexing operations and
 429          * installs appropriate dt_index_operations vector on success.
 430          *
 431          * Also probes for features. Operation is successful if all required
 432          * features are supported.
 433          */
 434         int   (*do_index_try)(const struct lu_env *env,
 435                               struct dt_object *dt,
 436                               const struct dt_index_features *feat);
 437         /**
 438          * Add nlink of the object
 439          * precondition: dt_object_exists(dt);
 440          */
 441         int   (*do_declare_ref_add)(const struct lu_env *env,
 442                                     struct dt_object *dt, struct thandle *th);
 443         int   (*do_ref_add)(const struct lu_env *env,
 444                             struct dt_object *dt, struct thandle *th);
 445         /**
 446          * Del nlink of the object
 447          * precondition: dt_object_exists(dt);
 448          */
 449         int   (*do_declare_ref_del)(const struct lu_env *env,
 450                                     struct dt_object *dt, struct thandle *th);
 451         int   (*do_ref_del)(const struct lu_env *env,
 452                             struct dt_object *dt, struct thandle *th);
 453
 454         struct obd_capa *(*do_capa_get)(const struct lu_env *env,
 455                                         struct dt_object *dt,
 456                                         struct lustre_capa *old,
 457                                         __u64 opc);
 458         int (*do_object_sync)(const struct lu_env *env, struct dt_object *obj,
 459                               __u64 start, __u64 end);
 460         /**
 461          * Get object info of next level. Currently, only get inode from osd.
 462          * This is only used by quota b=16542
 463          * precondition: dt_object_exists(dt);
 464          */
 465         int (*do_data_get)(const struct lu_env *env, struct dt_object *dt,
 466                            void **data);
 467
 468         /**
 469          * Lock object.
 470          */
 471         int (*do_object_lock)(const struct lu_env *env, struct dt_object *dt,
 472                               struct lustre_handle *lh,
 473                               struct ldlm_enqueue_info *einfo,
 474                               union ldlm_policy_data *policy);
 475
 476         int (*do_object_unlock)(const struct lu_env *env, struct dt_object *dt,
 477                                 struct ldlm_enqueue_info *einfo,
 478                                 union ldlm_policy_data *policy);
 479 };
 480
 481 /**
 482  * Per-dt-object operations on "file body".
 483  */
 484 struct dt_body_operations {
 485         /**
 486          * precondition: dt_object_exists(dt);
 487          */
 488         ssize_t (*dbo_read)(const struct lu_env *env, struct dt_object *dt,
 489                             struct lu_buf *buf, loff_t *pos,
 490                             struct lustre_capa *capa);
 491         /**
 492          * precondition: dt_object_exists(dt);
 493          */
 494         ssize_t (*dbo_declare_write)(const struct lu_env *env,
 495                                      struct dt_object *dt,
 496                                      const struct lu_buf *buf, loff_t pos,
 497                                      struct thandle *handle);
 498         ssize_t (*dbo_write)(const struct lu_env *env, struct dt_object *dt,
 499                              const struct lu_buf *buf, loff_t *pos,
 500                              struct thandle *handle, struct lustre_capa *capa,
 501                              int ignore_quota);
 502         /*
 503          * methods for zero-copy IO
 504          */
 505
 506         /*
 507          * precondition: dt_object_exists(dt);
 508          * returns:
 509          * < 0 - error code
 510          * = 0 - illegal
 511          * > 0 - number of local buffers prepared
 512          */
 513         int (*dbo_bufs_get)(const struct lu_env *env, struct dt_object *dt,
 514                             loff_t pos, ssize_t len, struct niobuf_local *lb,
 515                             int rw, struct lustre_capa *capa);
 516         /*
 517          * precondition: dt_object_exists(dt);
 518          */
 519         int (*dbo_bufs_put)(const struct lu_env *env, struct dt_object *dt,
 520                             struct niobuf_local *lb, int nr);
 521         /*
 522          * precondition: dt_object_exists(dt);
 523          */
 524         int (*dbo_write_prep)(const struct lu_env *env, struct dt_object *dt,
 525                               struct niobuf_local *lb, int nr);
 526         /*
 527          * precondition: dt_object_exists(dt);
 528          */
 529         int (*dbo_declare_write_commit)(const struct lu_env *env,
 530                                         struct dt_object *dt,
 531                                         struct niobuf_local *,
 532                                         int, struct thandle *);
 533         /*
 534          * precondition: dt_object_exists(dt);
 535          */
 536         int (*dbo_write_commit)(const struct lu_env *env, struct dt_object *dt,
 537                                 struct niobuf_local *, int, struct thandle *);
 538         /*
 539          * precondition: dt_object_exists(dt);
 540          */
 541         int (*dbo_read_prep)(const struct lu_env *env, struct dt_object *dt,
 542                              struct niobuf_local *lnb, int nr);
 543         int (*dbo_fiemap_get)(const struct lu_env *env, struct dt_object *dt,
 544                               struct ll_user_fiemap *fm);
 545         /**
 546          * Punch object's content
 547          * precondition: regular object, not index
 548          */
 549         int   (*dbo_declare_punch)(const struct lu_env *, struct dt_object *,
 550                                   __u64, __u64, struct thandle *th);
 551         int   (*dbo_punch)(const struct lu_env *env, struct dt_object *dt,
 552                           __u64 start, __u64 end, struct thandle *th,
 553                           struct lustre_capa *capa);
 554 };
 555
 556 /**
 557  * Incomplete type of index record.
 558  */
 559 struct dt_rec;
 560
 561 /**
 562  * Incomplete type of index key.
 563  */
 564 struct dt_key;
 565
 566 /**
 567  * Incomplete type of dt iterator.
 568  */
 569 struct dt_it;
 570
 571 /**
 572  * Per-dt-object operations on object as index.
 573  */
 574 struct dt_index_operations {
 575         /**
 576          * precondition: dt_object_exists(dt);
 577          */
 578         int (*dio_lookup)(const struct lu_env *env, struct dt_object *dt,
 579                           struct dt_rec *rec, const struct dt_key *key,
 580                           struct lustre_capa *capa);
 581         /**
 582          * precondition: dt_object_exists(dt);
 583          */
 584         int (*dio_declare_insert)(const struct lu_env *env,
 585                                   struct dt_object *dt,
 586                                   const struct dt_rec *rec,
 587                                   const struct dt_key *key,
 588                                   struct thandle *handle);
 589         int (*dio_insert)(const struct lu_env *env, struct dt_object *dt,
 590                           const struct dt_rec *rec, const struct dt_key *key,
 591                           struct thandle *handle, struct lustre_capa *capa,
 592                           int ignore_quota);
 593         /**
 594          * precondition: dt_object_exists(dt);
 595          */
 596         int (*dio_declare_delete)(const struct lu_env *env,
 597                                   struct dt_object *dt,
 598                                   const struct dt_key *key,
 599                                   struct thandle *handle);
 600         int (*dio_delete)(const struct lu_env *env, struct dt_object *dt,
 601                           const struct dt_key *key, struct thandle *handle,
 602                           struct lustre_capa *capa);
 603         /**
 604          * Iterator interface
 605          */
 606         struct dt_it_ops {
 607                 /**
 608                  * Allocate and initialize new iterator.
 609                  *
 610                  * precondition: dt_object_exists(dt);
 611                  */
 612                 struct dt_it *(*init)(const struct lu_env *env,
 613                                       struct dt_object *dt,
 614                                       __u32 attr,
 615                                       struct lustre_capa *capa);
 616                 void          (*fini)(const struct lu_env *env,
 617                                       struct dt_it *di);
 618                 int            (*get)(const struct lu_env *env,
 619                                       struct dt_it *di,
 620                                       const struct dt_key *key);
 621                 void           (*put)(const struct lu_env *env,
 622                                       struct dt_it *di);
 623                 int           (*next)(const struct lu_env *env,
 624                                       struct dt_it *di);
 625                 struct dt_key *(*key)(const struct lu_env *env,
 626                                       const struct dt_it *di);
 627                 int       (*key_size)(const struct lu_env *env,
 628                                       const struct dt_it *di);
 629                 int            (*rec)(const struct lu_env *env,
 630                                       const struct dt_it *di,
 631                                       struct dt_rec *rec,
 632                                       __u32 attr);
 633                 int        (*rec_size)(const struct lu_env *env,
 634                                        const struct dt_it *di,
 635                                       __u32 attr);
 636                 __u64        (*store)(const struct lu_env *env,
 637                                       const struct dt_it *di);
 638                 int           (*load)(const struct lu_env *env,
 639                                       const struct dt_it *di, __u64 hash);
 640                 int        (*key_rec)(const struct lu_env *env,
 641                                       const struct dt_it *di, void* key_rec);
 642         } dio_it;
 643 };
 644
 645 enum dt_otable_it_valid {
 646         DOIV_ERROR_HANDLE       = 0x0001,
 647         DOIV_DRYRUN             = 0x0002,
 648 };
 649
 650 enum dt_otable_it_flags {
 651         /* Exit when fail. */
 652         DOIF_FAILOUT    = 0x0001,
 653
 654         /* Reset iteration position to the device beginning. */
 655         DOIF_RESET      = 0x0002,
 656
 657         /* There is up layer component uses the iteration. */
 658         DOIF_OUTUSED    = 0x0004,
 659
 660         /* Check only without repairing. */
 661         DOIF_DRYRUN     = 0x0008,
 662 };
 663
 664 /* otable based iteration needs to use the common DT interation APIs.
 665  * To initialize the iteration, it needs call dio_it::init() firstly.
 666  * Here is how the otable based iteration should prepare arguments to
 667  * call dt_it_ops::init().
 668  *
 669  * For otable based iteration, the 32-bits 'attr' for dt_it_ops::init()
 670  * is composed of two parts:
 671  * low 16-bits is for valid bits, high 16-bits is for flags bits. */
 672 #define DT_OTABLE_IT_FLAGS_SHIFT        16
 673 #define DT_OTABLE_IT_FLAGS_MASK         0xffff0000
 674
 675 struct dt_device {
 676         struct lu_device                   dd_lu_dev;
 677         const struct dt_device_operations *dd_ops;
 678
 679         /**
 680          * List of dt_txn_callback (see below). This is not protected in any
 681          * way, because callbacks are supposed to be added/deleted only during
 682          * single-threaded start-up shut-down procedures.
 683          */
 684         struct list_head                   dd_txn_callbacks;
 685         unsigned int                       dd_record_fid_accessed:1;
 686 };
 687
 688 int  dt_device_init(struct dt_device *dev, struct lu_device_type *t);
 689 void dt_device_fini(struct dt_device *dev);
 690
 691 static inline int lu_device_is_dt(const struct lu_device *d)
 692 {
 693         return ergo(d != NULL, d->ld_type->ldt_tags & LU_DEVICE_DT);
 694 }
 695
 696 static inline struct dt_device * lu2dt_dev(struct lu_device *l)
 697 {
 698         LASSERT(lu_device_is_dt(l));
 699         return container_of0(l, struct dt_device, dd_lu_dev);
 700 }
 701
 702 struct dt_object {
 703         struct lu_object                   do_lu;
 704         const struct dt_object_operations *do_ops;
 705         const struct dt_body_operations   *do_body_ops;
 706         const struct dt_index_operations  *do_index_ops;
 707 };
 708
 709 /*
 710  * In-core representation of per-device local object OID storage
 711  */
 712 struct local_oid_storage {
 713         /* all initialized llog systems on this node linked by this */
 714         struct list_head  los_list;
 715
 716         /* how many handle's reference this los has */
 717         atomic_t          los_refcount;
 718         struct dt_device *los_dev;
 719         struct dt_object *los_obj;
 720
 721         /* data used to generate new fids */
 722         struct mutex      los_id_lock;
 723         __u64             los_seq;
 724         __u32             los_last_oid;
 725 };
 726
 727 static inline struct lu_device *dt2lu_dev(struct dt_device *d)
 728 {
 729         return &d->dd_lu_dev;
 730 }
 731
 732 static inline struct dt_object *lu2dt(struct lu_object *l)
 733 {
 734         LASSERT(l == NULL || IS_ERR(l) || lu_device_is_dt(l->lo_dev));
 735         return container_of0(l, struct dt_object, do_lu);
 736 }
 737
 738 int  dt_object_init(struct dt_object *obj,
 739                     struct lu_object_header *h, struct lu_device *d);
 740
 741 void dt_object_fini(struct dt_object *obj);
 742
 743 static inline int dt_object_exists(const struct dt_object *dt)
 744 {
 745         return lu_object_exists(&dt->do_lu);
 746 }
 747
 748 static inline int dt_object_remote(const struct dt_object *dt)
 749 {
 750         return lu_object_remote(&dt->do_lu);
 751 }
 752
 753 static inline struct dt_object *lu2dt_obj(struct lu_object *o)
 754 {
 755         LASSERT(ergo(o != NULL, lu_device_is_dt(o->lo_dev)));
 756         return container_of0(o, struct dt_object, do_lu);
 757 }
 758
 759 struct thandle_update {
 760         /* In DNE, one transaction can be disassembled into
 761          * updates on several different MDTs, and these updates
 762          * will be attached to tu_remote_update_list per target.
 763          * Only single thread will access the list, no need lock
 764          */
 765         struct list_head        tu_remote_update_list;
 766
 767         /* sent after or before local transaction */
 768         unsigned int            tu_sent_after_local_trans:1,
 769                                 tu_only_remote_trans:1;
 770 };
 771
 772 /**
 773  * This is the general purpose transaction handle.
 774  * 1. Transaction Life Cycle
 775  *      This transaction handle is allocated upon starting a new transaction,
 776  *      and deallocated after this transaction is committed.
 777  * 2. Transaction Nesting
 778  *      We do _NOT_ support nested transaction. So, every thread should only
 779  *      have one active transaction, and a transaction only belongs to one
 780  *      thread. Due to this, transaction handle need no reference count.
 781  * 3. Transaction & dt_object locking
 782  *      dt_object locks should be taken inside transaction.
 783  * 4. Transaction & RPC
 784  *      No RPC request should be issued inside transaction.
 785  */
 786 struct thandle {
 787         /** the dt device on which the transactions are executed */
 788         struct dt_device *th_dev;
 789
 790         atomic_t        th_refc;
 791         /* the size of transaction */
 792         int             th_alloc_size;
 793
 794         /** context for this transaction, tag is LCT_TX_HANDLE */
 795         struct lu_context th_ctx;
 796
 797         /** additional tags (layers can add in declare) */
 798         __u32             th_tags;
 799
 800         /** the last operation result in this transaction.
 801          * this value is used in recovery */
 802         __s32             th_result;
 803
 804         /** whether we need sync commit */
 805         unsigned int            th_sync:1;
 806
 807         /* local transation, no need to inform other layers */
 808         unsigned int            th_local:1;
 809
 810         struct thandle_update   *th_update;
 811 };
 812
 813 static inline void thandle_get(struct thandle *thandle)
 814 {
 815         atomic_inc(&thandle->th_refc);
 816 }
 817
 818 static inline void thandle_put(struct thandle *thandle)
 819 {
 820         if (atomic_dec_and_test(&thandle->th_refc)) {
 821                 if (thandle->th_update != NULL)
 822                         OBD_FREE_PTR(thandle->th_update);
 823                 OBD_FREE(thandle, thandle->th_alloc_size);
 824         }
 825 }
 826 /**
 827  * Transaction call-backs.
 828  *
 829  * These are invoked by osd (or underlying transaction engine) when
 830  * transaction changes state.
 831  *
 832  * Call-backs are used by upper layers to modify transaction parameters and to
 833  * perform some actions on for each transaction state transition. Typical
 834  * example is mdt registering call-back to write into last-received file
 835  * before each transaction commit.
 836  */
 837 struct dt_txn_callback {
 838         int (*dtc_txn_start)(const struct lu_env *env,
 839                              struct thandle *txn, void *cookie);
 840         int (*dtc_txn_stop)(const struct lu_env *env,
 841                             struct thandle *txn, void *cookie);
 842         void (*dtc_txn_commit)(struct thandle *txn, void *cookie);
 843         void                    *dtc_cookie;
 844         __u32                   dtc_tag;
 845         struct list_head        dtc_linkage;
 846 };
 847
 848 void dt_txn_callback_add(struct dt_device *dev, struct dt_txn_callback *cb);
 849 void dt_txn_callback_del(struct dt_device *dev, struct dt_txn_callback *cb);
 850
 851 int dt_txn_hook_start(const struct lu_env *env,
 852                       struct dt_device *dev, struct thandle *txn);
 853 int dt_txn_hook_stop(const struct lu_env *env, struct thandle *txn);
 854 void dt_txn_hook_commit(struct thandle *txn);
 855
 856 int dt_try_as_dir(const struct lu_env *env, struct dt_object *obj);
 857
 858 /**
 859  * Callback function used for parsing path.
 860  * \see llo_store_resolve
 861  */
 862 typedef int (*dt_entry_func_t)(const struct lu_env *env,
 863                             const char *name,
 864                             void *pvt);
 865
 866 #define DT_MAX_PATH 1024
 867
 868 int dt_path_parser(const struct lu_env *env,
 869                    char *local, dt_entry_func_t entry_func,
 870                    void *data);
 871
 872 struct dt_object *
 873 dt_store_resolve(const struct lu_env *env, struct dt_device *dt,
 874                  const char *path, struct lu_fid *fid);
 875
 876 struct dt_object *dt_store_open(const struct lu_env *env,
 877                                 struct dt_device *dt,
 878                                 const char *dirname,
 879                                 const char *filename,
 880                                 struct lu_fid *fid);
 881
 882 struct dt_object *dt_find_or_create(const struct lu_env *env,
 883                                     struct dt_device *dt,
 884                                     const struct lu_fid *fid,
 885                                     struct dt_object_format *dof,
 886                                     struct lu_attr *attr);
 887
 888 struct dt_object *dt_locate_at(const struct lu_env *env,
 889                                struct dt_device *dev,
 890                                const struct lu_fid *fid,
 891                                struct lu_device *top_dev,
 892                                const struct lu_object_conf *conf);
 893
 894 static inline struct dt_object *
 895 dt_locate(const struct lu_env *env, struct dt_device *dev,
 896           const struct lu_fid *fid)
 897 {
 898         return dt_locate_at(env, dev, fid,
 899                             dev->dd_lu_dev.ld_site->ls_top_dev, NULL);
 900 }
 901
 902 int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
 903                            const struct lu_fid *first_fid,
 904                            struct local_oid_storage **los);
 905 void local_oid_storage_fini(const struct lu_env *env,
 906                             struct local_oid_storage *los);
 907 int local_object_fid_generate(const struct lu_env *env,
 908                               struct local_oid_storage *los,
 909                               struct lu_fid *fid);
 910 int local_object_declare_create(const struct lu_env *env,
 911                                 struct local_oid_storage *los,
 912                                 struct dt_object *o,
 913                                 struct lu_attr *attr,
 914                                 struct dt_object_format *dof,
 915                                 struct thandle *th);
 916 int local_object_create(const struct lu_env *env,
 917                         struct local_oid_storage *los,
 918                         struct dt_object *o,
 919                         struct lu_attr *attr, struct dt_object_format *dof,
 920                         struct thandle *th);
 921 struct dt_object *local_file_find_or_create(const struct lu_env *env,
 922                                             struct local_oid_storage *los,
 923                                             struct dt_object *parent,
 924                                             const char *name, __u32 mode);
 925 struct dt_object *local_file_find_or_create_with_fid(const struct lu_env *env,
 926                                                      struct dt_device *dt,
 927                                                      const struct lu_fid *fid,
 928                                                      struct dt_object *parent,
 929                                                      const char *name,
 930                                                      __u32 mode);
 931 struct dt_object *
 932 local_index_find_or_create(const struct lu_env *env,
 933                            struct local_oid_storage *los,
 934                            struct dt_object *parent,
 935                            const char *name, __u32 mode,
 936                            const struct dt_index_features *ft);
 937 struct dt_object *
 938 local_index_find_or_create_with_fid(const struct lu_env *env,
 939                                     struct dt_device *dt,
 940                                     const struct lu_fid *fid,
 941                                     struct dt_object *parent,
 942                                     const char *name, __u32 mode,
 943                                     const struct dt_index_features *ft);
 944 int local_object_unlink(const struct lu_env *env, struct dt_device *dt,
 945                         struct dt_object *parent, const char *name);
 946
 947 static inline int dt_object_lock(const struct lu_env *env,
 948                                  struct dt_object *o, struct lustre_handle *lh,
 949                                  struct ldlm_enqueue_info *einfo,
 950                                  union ldlm_policy_data *policy)
 951 {
 952         LASSERT(o != NULL);
 953         LASSERT(o->do_ops != NULL);
 954         LASSERT(o->do_ops->do_object_lock != NULL);
 955         return o->do_ops->do_object_lock(env, o, lh, einfo, policy);
 956 }
 957
 958 static inline int dt_object_unlock(const struct lu_env *env,
 959                                    struct dt_object *o,
 960                                    struct ldlm_enqueue_info *einfo,
 961                                    union ldlm_policy_data *policy)
 962 {
 963         LASSERT(o != NULL);
 964         LASSERT(o->do_ops != NULL);
 965         LASSERT(o->do_ops->do_object_unlock != NULL);
 966         return o->do_ops->do_object_unlock(env, o, einfo, policy);
 967 }
 968
 969 int dt_lookup_dir(const struct lu_env *env, struct dt_object *dir,
 970                   const char *name, struct lu_fid *fid);
 971
 972 static inline int dt_object_sync(const struct lu_env *env, struct dt_object *o,
 973                                  __u64 start, __u64 end)
 974 {
 975         LASSERT(o);
 976         LASSERT(o->do_ops);
 977         LASSERT(o->do_ops->do_object_sync);
 978         return o->do_ops->do_object_sync(env, o, start, end);
 979 }
 980
 981 int dt_declare_version_set(const struct lu_env *env, struct dt_object *o,
 982                            struct thandle *th);
 983 void dt_version_set(const struct lu_env *env, struct dt_object *o,
 984                     dt_obj_version_t version, struct thandle *th);
 985 dt_obj_version_t dt_version_get(const struct lu_env *env, struct dt_object *o);
 986
 987
 988 int dt_read(const struct lu_env *env, struct dt_object *dt,
 989             struct lu_buf *buf, loff_t *pos);
 990 int dt_record_read(const struct lu_env *env, struct dt_object *dt,
 991                    struct lu_buf *buf, loff_t *pos);
 992 int dt_record_write(const struct lu_env *env, struct dt_object *dt,
 993                     const struct lu_buf *buf, loff_t *pos, struct thandle *th);
 994 typedef int (*dt_index_page_build_t)(const struct lu_env *env,
 995                                      union lu_page *lp, int nob,
 996                                      const struct dt_it_ops *iops,
 997                                      struct dt_it *it, __u32 attr, void *arg);
 998 int dt_index_walk(const struct lu_env *env, struct dt_object *obj,
 999                   const struct lu_rdpg *rdpg, dt_index_page_build_t filler,
1000                   void *arg);
1001 int dt_index_read(const struct lu_env *env, struct dt_device *dev,
1002                   struct idx_info *ii, const struct lu_rdpg *rdpg);
1003
1004 static inline struct thandle *dt_trans_create(const struct lu_env *env,
1005                                               struct dt_device *d)
1006 {
1007         LASSERT(d->dd_ops->dt_trans_create);
1008         return d->dd_ops->dt_trans_create(env, d);
1009 }
1010
1011 static inline int dt_trans_start(const struct lu_env *env,
1012                                  struct dt_device *d, struct thandle *th)
1013 {
1014         LASSERT(d->dd_ops->dt_trans_start);
1015         return d->dd_ops->dt_trans_start(env, d, th);
1016 }
1017
1018 /* for this transaction hooks shouldn't be called */
1019 static inline int dt_trans_start_local(const struct lu_env *env,
1020                                        struct dt_device *d, struct thandle *th)
1021 {
1022         LASSERT(d->dd_ops->dt_trans_start);
1023         th->th_local = 1;
1024         return d->dd_ops->dt_trans_start(env, d, th);
1025 }
1026
1027 static inline int dt_trans_stop(const struct lu_env *env,
1028                                 struct dt_device *d, struct thandle *th)
1029 {
1030         LASSERT(d->dd_ops->dt_trans_stop);
1031         return d->dd_ops->dt_trans_stop(env, d, th);
1032 }
1033
1034 static inline int dt_trans_cb_add(struct thandle *th,
1035                                   struct dt_txn_commit_cb *dcb)
1036 {
1037         LASSERT(th->th_dev->dd_ops->dt_trans_cb_add);
1038         dcb->dcb_magic = TRANS_COMMIT_CB_MAGIC;
1039         return th->th_dev->dd_ops->dt_trans_cb_add(th, dcb);
1040 }
1041 /** @} dt */
1042
1043
1044 static inline int dt_declare_record_write(const struct lu_env *env,
1045                                           struct dt_object *dt,
1046                                           const struct lu_buf *buf,
1047                                           loff_t pos,
1048                                           struct thandle *th)
1049 {
1050         int rc;
1051
1052         LASSERTF(dt != NULL, "dt is NULL when we want to write record\n");
1053         LASSERT(th != NULL);
1054         LASSERT(dt->do_body_ops);
1055         LASSERT(dt->do_body_ops->dbo_declare_write);
1056         rc = dt->do_body_ops->dbo_declare_write(env, dt, buf, pos, th);
1057         return rc;
1058 }
1059
1060 static inline int dt_declare_create(const struct lu_env *env,
1061                                     struct dt_object *dt,
1062                                     struct lu_attr *attr,
1063                                     struct dt_allocation_hint *hint,
1064                                     struct dt_object_format *dof,
1065                                     struct thandle *th)
1066 {
1067         LASSERT(dt);
1068         LASSERT(dt->do_ops);
1069         LASSERT(dt->do_ops->do_declare_create);
1070
1071         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_CREATE))
1072                 return cfs_fail_err;
1073
1074         return dt->do_ops->do_declare_create(env, dt, attr, hint, dof, th);
1075 }
1076
1077 static inline int dt_create(const struct lu_env *env,
1078                                     struct dt_object *dt,
1079                                     struct lu_attr *attr,
1080                                     struct dt_allocation_hint *hint,
1081                                     struct dt_object_format *dof,
1082                                     struct thandle *th)
1083 {
1084         LASSERT(dt);
1085         LASSERT(dt->do_ops);
1086         LASSERT(dt->do_ops->do_create);
1087
1088         if (CFS_FAULT_CHECK(OBD_FAIL_DT_CREATE))
1089                 return cfs_fail_err;
1090
1091         return dt->do_ops->do_create(env, dt, attr, hint, dof, th);
1092 }
1093
1094 static inline int dt_declare_destroy(const struct lu_env *env,
1095                                      struct dt_object *dt,
1096                                      struct thandle *th)
1097 {
1098         LASSERT(dt);
1099         LASSERT(dt->do_ops);
1100         LASSERT(dt->do_ops->do_declare_destroy);
1101
1102         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_DESTROY))
1103                 return cfs_fail_err;
1104
1105         return dt->do_ops->do_declare_destroy(env, dt, th);
1106 }
1107
1108 static inline int dt_destroy(const struct lu_env *env,
1109                              struct dt_object *dt,
1110                              struct thandle *th)
1111 {
1112         LASSERT(dt);
1113         LASSERT(dt->do_ops);
1114         LASSERT(dt->do_ops->do_destroy);
1115
1116         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DESTROY))
1117                 return cfs_fail_err;
1118
1119         return dt->do_ops->do_destroy(env, dt, th);
1120 }
1121
1122 static inline void dt_read_lock(const struct lu_env *env,
1123                                 struct dt_object *dt,
1124                                 unsigned role)
1125 {
1126         LASSERT(dt);
1127         LASSERT(dt->do_ops);
1128         LASSERT(dt->do_ops->do_read_lock);
1129         dt->do_ops->do_read_lock(env, dt, role);
1130 }
1131
1132 static inline void dt_write_lock(const struct lu_env *env,
1133                                 struct dt_object *dt,
1134                                 unsigned role)
1135 {
1136         LASSERT(dt);
1137         LASSERT(dt->do_ops);
1138         LASSERT(dt->do_ops->do_write_lock);
1139         dt->do_ops->do_write_lock(env, dt, role);
1140 }
1141
1142 static inline void dt_read_unlock(const struct lu_env *env,
1143                                 struct dt_object *dt)
1144 {
1145         LASSERT(dt);
1146         LASSERT(dt->do_ops);
1147         LASSERT(dt->do_ops->do_read_unlock);
1148         dt->do_ops->do_read_unlock(env, dt);
1149 }
1150
1151 static inline void dt_write_unlock(const struct lu_env *env,
1152                                 struct dt_object *dt)
1153 {
1154         LASSERT(dt);
1155         LASSERT(dt->do_ops);
1156         LASSERT(dt->do_ops->do_write_unlock);
1157         dt->do_ops->do_write_unlock(env, dt);
1158 }
1159
1160 static inline int dt_write_locked(const struct lu_env *env,
1161                                   struct dt_object *dt)
1162 {
1163         LASSERT(dt);
1164         LASSERT(dt->do_ops);
1165         LASSERT(dt->do_ops->do_write_locked);
1166         return dt->do_ops->do_write_locked(env, dt);
1167 }
1168
1169 static inline int dt_declare_attr_get(const struct lu_env *env,
1170                                       struct dt_object *dt,
1171                                       struct lustre_capa *capa)
1172 {
1173         LASSERT(dt);
1174         LASSERT(dt->do_ops);
1175         LASSERT(dt->do_ops->do_declare_attr_get);
1176
1177         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_ATTR_GET))
1178                 return cfs_fail_err;
1179
1180         return dt->do_ops->do_declare_attr_get(env, dt, capa);
1181 }
1182
1183 static inline int dt_attr_get(const struct lu_env *env, struct dt_object *dt,
1184                               struct lu_attr *la, void *arg)
1185 {
1186         LASSERT(dt);
1187         LASSERT(dt->do_ops);
1188         LASSERT(dt->do_ops->do_attr_get);
1189
1190         if (CFS_FAULT_CHECK(OBD_FAIL_DT_ATTR_GET))
1191                 return cfs_fail_err;
1192
1193         return dt->do_ops->do_attr_get(env, dt, la, arg);
1194 }
1195
1196 static inline int dt_declare_attr_set(const struct lu_env *env,
1197                                       struct dt_object *dt,
1198                                       const struct lu_attr *la,
1199                                       struct thandle *th)
1200 {
1201         LASSERT(dt);
1202         LASSERT(dt->do_ops);
1203         LASSERT(dt->do_ops->do_declare_attr_set);
1204
1205         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_ATTR_SET))
1206                 return cfs_fail_err;
1207
1208         return dt->do_ops->do_declare_attr_set(env, dt, la, th);
1209 }
1210
1211 static inline int dt_attr_set(const struct lu_env *env, struct dt_object *dt,
1212                               const struct lu_attr *la, struct thandle *th,
1213                               struct lustre_capa *capa)
1214 {
1215         LASSERT(dt);
1216         LASSERT(dt->do_ops);
1217         LASSERT(dt->do_ops->do_attr_set);
1218
1219         if (CFS_FAULT_CHECK(OBD_FAIL_DT_ATTR_SET))
1220                 return cfs_fail_err;
1221
1222         return dt->do_ops->do_attr_set(env, dt, la, th, capa);
1223 }
1224
1225 static inline int dt_declare_ref_add(const struct lu_env *env,
1226                                      struct dt_object *dt, struct thandle *th)
1227 {
1228         LASSERT(dt);
1229         LASSERT(dt->do_ops);
1230         LASSERT(dt->do_ops->do_declare_ref_add);
1231
1232         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_REF_ADD))
1233                 return cfs_fail_err;
1234
1235         return dt->do_ops->do_declare_ref_add(env, dt, th);
1236 }
1237
1238 static inline int dt_ref_add(const struct lu_env *env,
1239                              struct dt_object *dt, struct thandle *th)
1240 {
1241         LASSERT(dt);
1242         LASSERT(dt->do_ops);
1243         LASSERT(dt->do_ops->do_ref_add);
1244
1245         if (CFS_FAULT_CHECK(OBD_FAIL_DT_REF_ADD))
1246                 return cfs_fail_err;
1247
1248         return dt->do_ops->do_ref_add(env, dt, th);
1249 }
1250
1251 static inline int dt_declare_ref_del(const struct lu_env *env,
1252                                      struct dt_object *dt, struct thandle *th)
1253 {
1254         LASSERT(dt);
1255         LASSERT(dt->do_ops);
1256         LASSERT(dt->do_ops->do_declare_ref_del);
1257
1258         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_REF_DEL))
1259                 return cfs_fail_err;
1260
1261         return dt->do_ops->do_declare_ref_del(env, dt, th);
1262 }
1263
1264 static inline int dt_ref_del(const struct lu_env *env,
1265                              struct dt_object *dt, struct thandle *th)
1266 {
1267         LASSERT(dt);
1268         LASSERT(dt->do_ops);
1269         LASSERT(dt->do_ops->do_ref_del);
1270
1271         if (CFS_FAULT_CHECK(OBD_FAIL_DT_REF_DEL))
1272                 return cfs_fail_err;
1273
1274         return dt->do_ops->do_ref_del(env, dt, th);
1275 }
1276
1277 static inline struct obd_capa *dt_capa_get(const struct lu_env *env,
1278                                            struct dt_object *dt,
1279                                            struct lustre_capa *old, __u64 opc)
1280 {
1281         LASSERT(dt);
1282         LASSERT(dt->do_ops);
1283         LASSERT(dt->do_ops->do_capa_get);
1284         return dt->do_ops->do_capa_get(env, dt, old, opc);
1285 }
1286
1287 static inline int dt_bufs_get(const struct lu_env *env, struct dt_object *d,
1288                               struct niobuf_remote *rnb,
1289                               struct niobuf_local *lnb, int rw,
1290                               struct lustre_capa *capa)
1291 {
1292         LASSERT(d);
1293         LASSERT(d->do_body_ops);
1294         LASSERT(d->do_body_ops->dbo_bufs_get);
1295         return d->do_body_ops->dbo_bufs_get(env, d, rnb->rnb_offset,
1296                                             rnb->rnb_len, lnb, rw, capa);
1297 }
1298
1299 static inline int dt_bufs_put(const struct lu_env *env, struct dt_object *d,
1300                               struct niobuf_local *lnb, int n)
1301 {
1302         LASSERT(d);
1303         LASSERT(d->do_body_ops);
1304         LASSERT(d->do_body_ops->dbo_bufs_put);
1305         return d->do_body_ops->dbo_bufs_put(env, d, lnb, n);
1306 }
1307
1308 static inline int dt_write_prep(const struct lu_env *env, struct dt_object *d,
1309                                 struct niobuf_local *lnb, int n)
1310 {
1311         LASSERT(d);
1312         LASSERT(d->do_body_ops);
1313         LASSERT(d->do_body_ops->dbo_write_prep);
1314         return d->do_body_ops->dbo_write_prep(env, d, lnb, n);
1315 }
1316
1317 static inline int dt_declare_write_commit(const struct lu_env *env,
1318                                           struct dt_object *d,
1319                                           struct niobuf_local *lnb,
1320                                           int n, struct thandle *th)
1321 {
1322         LASSERTF(d != NULL, "dt is NULL when we want to declare write\n");
1323         LASSERT(th != NULL);
1324         return d->do_body_ops->dbo_declare_write_commit(env, d, lnb, n, th);
1325 }
1326
1327
1328 static inline int dt_write_commit(const struct lu_env *env,
1329                                   struct dt_object *d, struct niobuf_local *lnb,
1330                                   int n, struct thandle *th)
1331 {
1332         LASSERT(d);
1333         LASSERT(d->do_body_ops);
1334         LASSERT(d->do_body_ops->dbo_write_commit);
1335         return d->do_body_ops->dbo_write_commit(env, d, lnb, n, th);
1336 }
1337
1338 static inline int dt_read_prep(const struct lu_env *env, struct dt_object *d,
1339                                struct niobuf_local *lnb, int n)
1340 {
1341         LASSERT(d);
1342         LASSERT(d->do_body_ops);
1343         LASSERT(d->do_body_ops->dbo_read_prep);
1344         return d->do_body_ops->dbo_read_prep(env, d, lnb, n);
1345 }
1346
1347 static inline int dt_declare_punch(const struct lu_env *env,
1348                                    struct dt_object *dt, __u64 start,
1349                                    __u64 end, struct thandle *th)
1350 {
1351         LASSERT(dt);
1352         LASSERT(dt->do_body_ops);
1353         LASSERT(dt->do_body_ops->dbo_declare_punch);
1354         return dt->do_body_ops->dbo_declare_punch(env, dt, start, end, th);
1355 }
1356
1357 static inline int dt_punch(const struct lu_env *env, struct dt_object *dt,
1358                            __u64 start, __u64 end, struct thandle *th,
1359                            struct lustre_capa *capa)
1360 {
1361         LASSERT(dt);
1362         LASSERT(dt->do_body_ops);
1363         LASSERT(dt->do_body_ops->dbo_punch);
1364         return dt->do_body_ops->dbo_punch(env, dt, start, end, th, capa);
1365 }
1366
1367 static inline int dt_fiemap_get(const struct lu_env *env, struct dt_object *d,
1368                                 struct ll_user_fiemap *fm)
1369 {
1370         LASSERT(d);
1371         if (d->do_body_ops == NULL)
1372                 return -EPROTO;
1373         if (d->do_body_ops->dbo_fiemap_get == NULL)
1374                 return -EOPNOTSUPP;
1375         return d->do_body_ops->dbo_fiemap_get(env, d, fm);
1376 }
1377
1378 static inline int dt_statfs(const struct lu_env *env, struct dt_device *dev,
1379                             struct obd_statfs *osfs)
1380 {
1381         LASSERT(dev);
1382         LASSERT(dev->dd_ops);
1383         LASSERT(dev->dd_ops->dt_statfs);
1384         return dev->dd_ops->dt_statfs(env, dev, osfs);
1385 }
1386
1387 static inline int dt_root_get(const struct lu_env *env, struct dt_device *dev,
1388                               struct lu_fid *f)
1389 {
1390         LASSERT(dev);
1391         LASSERT(dev->dd_ops);
1392         LASSERT(dev->dd_ops->dt_root_get);
1393         return dev->dd_ops->dt_root_get(env, dev, f);
1394 }
1395
1396 static inline void dt_conf_get(const struct lu_env *env,
1397                                const struct dt_device *dev,
1398                                struct dt_device_param *param)
1399 {
1400         LASSERT(dev);
1401         LASSERT(dev->dd_ops);
1402         LASSERT(dev->dd_ops->dt_conf_get);
1403         return dev->dd_ops->dt_conf_get(env, dev, param);
1404 }
1405
1406 static inline int dt_sync(const struct lu_env *env, struct dt_device *dev)
1407 {
1408         LASSERT(dev);
1409         LASSERT(dev->dd_ops);
1410         LASSERT(dev->dd_ops->dt_sync);
1411         return dev->dd_ops->dt_sync(env, dev);
1412 }
1413
1414 static inline int dt_ro(const struct lu_env *env, struct dt_device *dev)
1415 {
1416         LASSERT(dev);
1417         LASSERT(dev->dd_ops);
1418         LASSERT(dev->dd_ops->dt_ro);
1419         return dev->dd_ops->dt_ro(env, dev);
1420 }
1421
1422 static inline int dt_declare_insert(const struct lu_env *env,
1423                                     struct dt_object *dt,
1424                                     const struct dt_rec *rec,
1425                                     const struct dt_key *key,
1426                                     struct thandle *th)
1427 {
1428         LASSERT(dt);
1429         LASSERT(dt->do_index_ops);
1430         LASSERT(dt->do_index_ops->dio_declare_insert);
1431
1432         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_INSERT))
1433                 return cfs_fail_err;
1434
1435         return dt->do_index_ops->dio_declare_insert(env, dt, rec, key, th);
1436 }
1437
1438 static inline int dt_insert(const struct lu_env *env,
1439                                     struct dt_object *dt,
1440                                     const struct dt_rec *rec,
1441                                     const struct dt_key *key,
1442                                     struct thandle *th,
1443                                     struct lustre_capa *capa,
1444                                     int noquota)
1445 {
1446         LASSERT(dt);
1447         LASSERT(dt->do_index_ops);
1448         LASSERT(dt->do_index_ops->dio_insert);
1449
1450         if (CFS_FAULT_CHECK(OBD_FAIL_DT_INSERT))
1451                 return cfs_fail_err;
1452
1453         return dt->do_index_ops->dio_insert(env, dt, rec, key, th,
1454                                             capa, noquota);
1455 }
1456
1457 static inline int dt_declare_xattr_del(const struct lu_env *env,
1458                                        struct dt_object *dt,
1459                                        const char *name,
1460                                        struct thandle *th)
1461 {
1462         LASSERT(dt);
1463         LASSERT(dt->do_ops);
1464         LASSERT(dt->do_ops->do_declare_xattr_del);
1465
1466         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_XATTR_DEL))
1467                 return cfs_fail_err;
1468
1469         return dt->do_ops->do_declare_xattr_del(env, dt, name, th);
1470 }
1471
1472 static inline int dt_xattr_del(const struct lu_env *env,
1473                                struct dt_object *dt, const char *name,
1474                                struct thandle *th,
1475                                struct lustre_capa *capa)
1476 {
1477         LASSERT(dt);
1478         LASSERT(dt->do_ops);
1479         LASSERT(dt->do_ops->do_xattr_del);
1480
1481         if (CFS_FAULT_CHECK(OBD_FAIL_DT_XATTR_DEL))
1482                 return cfs_fail_err;
1483
1484         return dt->do_ops->do_xattr_del(env, dt, name, th, capa);
1485 }
1486
1487 static inline int dt_declare_xattr_set(const struct lu_env *env,
1488                                       struct dt_object *dt,
1489                                       const struct lu_buf *buf,
1490                                       const char *name, int fl,
1491                                       struct thandle *th)
1492 {
1493         LASSERT(dt);
1494         LASSERT(dt->do_ops);
1495         LASSERT(dt->do_ops->do_declare_xattr_set);
1496
1497         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_XATTR_SET))
1498                 return cfs_fail_err;
1499
1500         return dt->do_ops->do_declare_xattr_set(env, dt, buf, name, fl, th);
1501 }
1502
1503 static inline int dt_xattr_set(const struct lu_env *env,
1504                               struct dt_object *dt, const struct lu_buf *buf,
1505                               const char *name, int fl, struct thandle *th,
1506                               struct lustre_capa *capa)
1507 {
1508         LASSERT(dt);
1509         LASSERT(dt->do_ops);
1510         LASSERT(dt->do_ops->do_xattr_set);
1511
1512         if (CFS_FAULT_CHECK(OBD_FAIL_DT_XATTR_SET))
1513                 return cfs_fail_err;
1514
1515         return dt->do_ops->do_xattr_set(env, dt, buf, name, fl, th, capa);
1516 }
1517
1518 static inline int dt_declare_xattr_get(const struct lu_env *env,
1519                                        struct dt_object *dt,
1520                                        struct lu_buf *buf,
1521                                        const char *name,
1522                                        struct lustre_capa *capa)
1523 {
1524         LASSERT(dt);
1525         LASSERT(dt->do_ops);
1526         LASSERT(dt->do_ops->do_declare_xattr_get);
1527
1528         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_XATTR_GET))
1529                 return cfs_fail_err;
1530
1531         return dt->do_ops->do_declare_xattr_get(env, dt, buf, name, capa);
1532 }
1533
1534 static inline int dt_xattr_get(const struct lu_env *env,
1535                               struct dt_object *dt, struct lu_buf *buf,
1536                               const char *name, struct lustre_capa *capa)
1537 {
1538         LASSERT(dt);
1539         LASSERT(dt->do_ops);
1540         LASSERT(dt->do_ops->do_xattr_get);
1541
1542         if (CFS_FAULT_CHECK(OBD_FAIL_DT_XATTR_GET))
1543                 return cfs_fail_err;
1544
1545         return dt->do_ops->do_xattr_get(env, dt, buf, name, capa);
1546 }
1547
1548 static inline int dt_xattr_list(const struct lu_env *env,
1549                                struct dt_object *dt, struct lu_buf *buf,
1550                                struct lustre_capa *capa)
1551 {
1552         LASSERT(dt);
1553         LASSERT(dt->do_ops);
1554         LASSERT(dt->do_ops->do_xattr_list);
1555
1556         if (CFS_FAULT_CHECK(OBD_FAIL_DT_XATTR_LIST))
1557                 return cfs_fail_err;
1558
1559         return dt->do_ops->do_xattr_list(env, dt, buf, capa);
1560 }
1561
1562 static inline int dt_declare_delete(const struct lu_env *env,
1563                                     struct dt_object *dt,
1564                                     const struct dt_key *key,
1565                                     struct thandle *th)
1566 {
1567         LASSERT(dt);
1568         LASSERT(dt->do_index_ops);
1569         LASSERT(dt->do_index_ops->dio_declare_delete);
1570
1571         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DECLARE_DELETE))
1572                 return cfs_fail_err;
1573
1574         return dt->do_index_ops->dio_declare_delete(env, dt, key, th);
1575 }
1576
1577 static inline int dt_delete(const struct lu_env *env,
1578                             struct dt_object *dt,
1579                             const struct dt_key *key,
1580                             struct thandle *th,
1581                             struct lustre_capa *capa)
1582 {
1583         LASSERT(dt);
1584         LASSERT(dt->do_index_ops);
1585         LASSERT(dt->do_index_ops->dio_delete);
1586
1587         if (CFS_FAULT_CHECK(OBD_FAIL_DT_DELETE))
1588                 return cfs_fail_err;
1589
1590         return dt->do_index_ops->dio_delete(env, dt, key, th, capa);
1591 }
1592
1593 static inline int dt_commit_async(const struct lu_env *env,
1594                                   struct dt_device *dev)
1595 {
1596         LASSERT(dev);
1597         LASSERT(dev->dd_ops);
1598         LASSERT(dev->dd_ops->dt_commit_async);
1599         return dev->dd_ops->dt_commit_async(env, dev);
1600 }
1601
1602 static inline int dt_init_capa_ctxt(const struct lu_env *env,
1603                                     struct dt_device *dev,
1604                                     int mode, unsigned long timeout,
1605                                     __u32 alg, struct lustre_capa_key *keys)
1606 {
1607         LASSERT(dev);
1608         LASSERT(dev->dd_ops);
1609         LASSERT(dev->dd_ops->dt_init_capa_ctxt);
1610         return dev->dd_ops->dt_init_capa_ctxt(env, dev, mode,
1611                                               timeout, alg, keys);
1612 }
1613
1614 static inline int dt_lookup(const struct lu_env *env,
1615                             struct dt_object *dt,
1616                             struct dt_rec *rec,
1617                             const struct dt_key *key,
1618                             struct lustre_capa *capa)
1619 {
1620         int ret;
1621
1622         LASSERT(dt);
1623         LASSERT(dt->do_index_ops);
1624         LASSERT(dt->do_index_ops->dio_lookup);
1625
1626         if (CFS_FAULT_CHECK(OBD_FAIL_DT_LOOKUP))
1627                 return cfs_fail_err;
1628
1629         ret = dt->do_index_ops->dio_lookup(env, dt, rec, key, capa);
1630         if (ret > 0)
1631                 ret = 0;
1632         else if (ret == 0)
1633                 ret = -ENOENT;
1634         return ret;
1635 }
1636
1637 #define LU221_BAD_TIME (0x80000000U + 24 * 3600)
1638
1639 struct dt_find_hint {
1640         struct lu_fid        *dfh_fid;
1641         struct dt_device     *dfh_dt;
1642         struct dt_object     *dfh_o;
1643 };
1644
1645 struct dt_insert_rec {
1646         union {
1647                 const struct lu_fid     *rec_fid;
1648                 void                    *rec_data;
1649         };
1650         union {
1651                 struct {
1652                         __u32            rec_type;
1653                         __u32            rec_padding;
1654                 };
1655                 __u64                    rec_misc;
1656         };
1657 };
1658
1659 struct dt_thread_info {
1660         char                     dti_buf[DT_MAX_PATH];
1661         struct dt_find_hint      dti_dfh;
1662         struct lu_attr           dti_attr;
1663         struct lu_fid            dti_fid;
1664         struct dt_object_format  dti_dof;
1665         struct lustre_mdt_attrs  dti_lma;
1666         struct lu_buf            dti_lb;
1667         struct lu_object_conf    dti_conf;
1668         loff_t                   dti_off;
1669         struct dt_insert_rec     dti_dt_rec;
1670 };
1671
1672 extern struct lu_context_key dt_key;
1673
1674 static inline struct dt_thread_info *dt_info(const struct lu_env *env)
1675 {
1676         struct dt_thread_info *dti;
1677
1678         dti = lu_context_key_get(&env->le_ctx, &dt_key);
1679         LASSERT(dti);
1680         return dti;
1681 }
1682
1683 int dt_global_init(void);
1684 void dt_global_fini(void);
1685
1686 # ifdef LPROCFS
1687 #ifndef HAVE_ONLY_PROCFS_SEQ
1688 int lprocfs_dt_rd_blksize(char *page, char **start, off_t off,
1689                           int count, int *eof, void *data);
1690 int lprocfs_dt_rd_kbytestotal(char *page, char **start, off_t off,
1691                               int count, int *eof, void *data);
1692 int lprocfs_dt_rd_kbytesfree(char *page, char **start, off_t off,
1693                              int count, int *eof, void *data);
1694 int lprocfs_dt_rd_kbytesavail(char *page, char **start, off_t off,
1695                               int count, int *eof, void *data);
1696 int lprocfs_dt_rd_filestotal(char *page, char **start, off_t off,
1697                              int count, int *eof, void *data);
1698 int lprocfs_dt_rd_filesfree(char *page, char **start, off_t off,
1699                             int count, int *eof, void *data);
1700 #endif
1701 int lprocfs_dt_blksize_seq_show(struct seq_file *m, void *v);
1702 int lprocfs_dt_kbytestotal_seq_show(struct seq_file *m, void *v);
1703 int lprocfs_dt_kbytesfree_seq_show(struct seq_file *m, void *v);
1704 int lprocfs_dt_kbytesavail_seq_show(struct seq_file *m, void *v);
1705 int lprocfs_dt_filestotal_seq_show(struct seq_file *m, void *v);
1706 int lprocfs_dt_filesfree_seq_show(struct seq_file *m, void *v);
1707 # endif /* LPROCFS */
1708
1709 #endif /* __LUSTRE_DT_OBJECT_H */