lustre/include/obd_class.h

   1 // SPDX-License-Identifier: GPL-2.0
   2
   3 /*
   4  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
   5  * Use is subject to license terms.
   6  *
   7  * Copyright (c) 2011, 2017, Intel Corporation.
   8  */
   9
  10 /*
  11  * This file is part of Lustre, http://www.lustre.org/
  12  *
  13  * lustre/include/obd_class.h
  14  *
  15  * Header defining common operations on OBD devices.
  16  *
  17  */
  18
  19 #ifndef __CLASS_OBD_H
  20 #define __CLASS_OBD_H
  21
  22 #include <linux/kobject.h>
  23 #include <obd_support.h>
  24 #include <lustre_import.h>
  25 #include <lustre_net.h>
  26 #include <obd.h>
  27 #include <lustre_lib.h>
  28 #include <uapi/linux/lustre/lustre_idl.h>
  29 #include <lprocfs_status.h>
  30 #ifdef HAVE_SERVER_SUPPORT
  31 #include <lu_target.h>
  32 #include <obd_target.h>
  33 #include <dt_object.h>
  34 #endif
  35
  36 #define OBD_STATFS_NODELAY      0x0001  /* requests should be send without delay
  37                                          * and resends for avoid deadlocks */
  38 #define OBD_STATFS_FROM_CACHE   0x0002  /* the statfs callback should not update
  39                                          * obd_osfs_age */
  40 #define OBD_STATFS_FOR_MDT0     0x0004  /* The statfs is only for retrieving
  41                                          * information from MDT0. */
  42 #define OBD_STATFS_SUM          0x0008  /* get aggregated statfs from MDT */
  43 #define OBD_STATFS_NESTED       0x0010  /* Call while already holding
  44                                          * obd_dev_mutex of a difference
  45                                          * device.
  46                                          */
  47
  48 #define OBD_MAX_INDEX xa_limit_31b.max
  49
  50 #define obd_device_find(devno)                                          \
  51         xa_find(&obd_devs, &devno, OBD_MAX_INDEX, XA_PRESENT)
  52
  53 #define obd_device_find_after(devno)                                    \
  54         xa_find_after(&obd_devs, &devno, OBD_MAX_INDEX, XA_PRESENT)
  55
  56 #define obd_device_for_each(devno, obd)                 \
  57         xa_for_each(&obd_devs, devno, obd)
  58
  59 #define obd_device_for_each_start(devno, obd, start)    \
  60         xa_for_each_start(&obd_devs, devno, obd, start)
  61
  62 #define obd_device_for_each_cond(devno, obd, cond)       \
  63         obd_device_for_each(devno, obd)                  \
  64         if (cond)
  65
  66 #define obd_device_for_each_uuid(devno, obd, uuid)       \
  67         obd_device_for_each_cond(devno, obd,             \
  68                                  obd_uuid_equals(uuid, &obd->obd_uuid))
  69
  70 #define obd_device_lock() xa_lock(&obd_devs)
  71 #define obd_device_unlock() xa_unlock(&obd_devs)
  72
  73 /* OBD Operations Declarations */
  74 extern struct obd_device *class_exp2obd(struct obd_export *);
  75 extern int class_handle_ioctl(unsigned int cmd, void __user *uarg);
  76 int lustre_get_jobid(char *jobid, size_t len);
  77 void lustre_jobid_clear(const char *jobid);
  78 void jobid_cache_fini(void);
  79 int jobid_cache_init(void);
  80 char *jobid_current(void);
  81 int jobid_set_current(char *jobid);
  82
  83 struct lu_device_type;
  84
  85 /* genops.c */
  86 extern struct xarray obd_devs;
  87 struct obd_export *class_conn2export(struct lustre_handle *);
  88 #ifdef HAVE_SERVER_SUPPORT
  89 struct obd_type *class_add_symlinks(const char *name, bool enable_proc);
  90 #endif
  91 int class_register_type(const struct obd_ops *dt_ops,
  92                         const struct md_ops *md_ops, bool enable_proc,
  93                         const char *nm, struct lu_device_type *ldt);
  94 int class_unregister_type(const char *nm);
  95
  96 struct obd_device *class_newdev(const char *type_name, const char *name,
  97                                 const char *uuid);
  98 int class_register_device(struct obd_device *obd);
  99 void class_unregister_device(struct obd_device *obd);
 100 void class_free_dev(struct obd_device *obd);
 101
 102 struct obd_device *class_str2obd(const char *str);
 103 int class_name2dev(const char *name);
 104 struct obd_device *class_name2obd(const char *name);
 105 int class_uuid2dev(struct obd_uuid *uuid);
 106 struct obd_device *class_uuid2obd(struct obd_uuid *uuid);
 107 struct obd_device *class_find_client_obd(struct obd_uuid *tgt_uuid,
 108                                          const char *type_name,
 109                                          struct obd_uuid *grp_uuid);
 110 struct obd_device *class_num2obd(int num);
 111 int class_obd_devs_count(void);
 112
 113 int class_notify_sptlrpc_conf(const char *fsname, int namelen);
 114
 115 static inline char *obd_export_nid2str(struct obd_export *exp)
 116 {
 117         return exp->exp_connection == NULL ?
 118                "<unknown>" : libcfs_nidstr(&exp->exp_connection->c_peer.nid);
 119 }
 120
 121 static inline char *obd_import_nid2str(struct obd_import *imp)
 122 {
 123         return imp->imp_connection == NULL ?
 124                "<unknown>" : libcfs_nidstr(&imp->imp_connection->c_peer.nid);
 125 }
 126
 127 int obd_export_evict_by_nid(struct obd_device *obd, const char *nid);
 128 int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid);
 129 int obd_connect_flags2str(char *page, int count, __u64 flags, __u64 flags2,
 130                           const char *sep);
 131
 132 int obd_zombie_impexp_init(void);
 133 void obd_zombie_impexp_stop(void);
 134 void obd_zombie_impexp_cull(void);
 135 void obd_zombie_barrier(void);
 136 void obd_exports_barrier(struct obd_device *obd);
 137 int kuc_len(int payload_len);
 138 struct kuc_hdr * kuc_ptr(void *p);
 139 void *kuc_alloc(int payload_len, int transport, int type);
 140 void kuc_free(void *p, int payload_len);
 141 int obd_get_request_slot(struct client_obd *cli);
 142 void obd_put_request_slot(struct client_obd *cli);
 143 __u32 obd_get_max_rpcs_in_flight(struct client_obd *cli);
 144 int obd_set_max_rpcs_in_flight(struct client_obd *cli, __u32 max);
 145 __u16 obd_get_max_mod_rpcs_in_flight(struct client_obd *cli);
 146 int obd_set_max_mod_rpcs_in_flight(struct client_obd *cli, __u16 max);
 147 int obd_mod_rpc_stats_seq_show(struct client_obd *cli, struct seq_file *seq);
 148
 149 __u16 obd_get_mod_rpc_slot(struct client_obd *cli, __u32 opc);
 150 void obd_put_mod_rpc_slot(struct client_obd *cli, __u32 opc, __u16 tag);
 151
 152 struct llog_handle;
 153 struct llog_rec_hdr;
 154 typedef int (*llog_cb_t)(const struct lu_env *, struct llog_handle *,
 155                          struct llog_rec_hdr *, void *);
 156
 157 struct obd_export *obd_stale_export_get(void);
 158 void obd_stale_export_put(struct obd_export *exp);
 159 void obd_stale_export_adjust(struct obd_export *exp);
 160
 161 /* obd_config.c */
 162 /* For interoperability */
 163 struct cfg_interop_param {
 164         char *old_param;
 165         char *new_param;
 166 };
 167
 168 char *lustre_cfg_string(struct lustre_cfg *lcfg, u32 index);
 169 struct lustre_cfg *lustre_cfg_rename(struct lustre_cfg *cfg,
 170                                      const char *new_name);
 171 void print_lustre_cfg(struct lustre_cfg *lcfg);
 172 int class_process_config(struct lustre_cfg *lcfg);
 173 ssize_t class_set_global(const char *param);
 174 ssize_t class_modify_config(struct lustre_cfg *lcfg, const char *prefix,
 175                             struct kobject *kobj);
 176 int class_attach(struct lustre_cfg *lcfg);
 177 int class_setup(struct obd_device *obd, struct lustre_cfg *lcfg);
 178 int class_cleanup(struct obd_device *obd, struct lustre_cfg *lcfg);
 179 int class_detach(struct obd_device *obd, struct lustre_cfg *lcfg);
 180
 181 int class_find_param(char *buf, char *key, char **valp);
 182 struct cfg_interop_param *class_find_old_param(const char *param,
 183                                                struct cfg_interop_param *ptr);
 184 int class_get_next_param(char **params, char *copy);
 185 int class_match_param(char *buf, const char *key, char **valp);
 186 int class_parse_nid(char *buf, struct lnet_nid *nid, char **endh);
 187 int class_parse_nid_quiet(char *buf, struct lnet_nid *nid, char **endh);
 188 int class_parse_net(char *buf, u32 *net, char **endh);
 189 int class_match_nid(char *buf, char *key, struct lnet_nid *nid);
 190 int class_match_net(char *buf, char *key, u32 net);
 191
 192 struct obd_device *class_incref(struct obd_device *obd,
 193                                 const char *scope, const void *source);
 194 void class_decref(struct obd_device *obd,
 195                   const char *scope, const void *source);
 196 void dump_exports(struct obd_device *obd, int locks, int debug_level);
 197 int class_config_llog_handler(const struct lu_env *env,
 198                               struct llog_handle *handle,
 199                               struct llog_rec_hdr *rec, void *data);
 200 int class_add_conn(struct obd_device *obd, struct lustre_cfg *lcfg);
 201
 202 #define CFG_F_START     0x01   /* Set when we start updating from a log */
 203 #define CFG_F_MARKER    0x02   /* We are within a maker */
 204 #define CFG_F_SKIP      0x04   /* We should ignore this cfg command */
 205 #define CFG_F_EXCLUDE   0x10   /* OST exclusion list */
 206
 207 /* Passed as data param to class_config_parse_llog */
 208 struct config_llog_instance {
 209         unsigned long            cfg_instance;
 210         struct super_block      *cfg_sb;
 211         struct obd_uuid          cfg_uuid;
 212         llog_cb_t                cfg_callback;
 213         int                      cfg_last_idx; /* for partial llog processing */
 214         int                      cfg_flags;
 215         __u32                    cfg_lwp_idx;
 216         __u32                    cfg_sub_clds;
 217 };
 218 int class_config_parse_llog(const struct lu_env *env, struct llog_ctxt *ctxt,
 219                             char *name, struct config_llog_instance *cfg);
 220
 221 /**
 222  * Generate a unique configuration instance for this mount
 223  *
 224  * Temporary hack to bypass ASLR in 4.15+ kernels, a better fix soon.
 225  * For now, use the same value as before - the superblock pointer value.
 226  *
 227  * Using the client UUID would be an option, but it needs more testing.
 228  */
 229 static inline unsigned long ll_get_cfg_instance(struct super_block *sb)
 230 {
 231         return (unsigned long)sb;
 232 }
 233
 234 #define CONFIG_SUB_SPTLRPC      0x01
 235 #define CONFIG_SUB_RECOVER      0x02
 236 #define CONFIG_SUB_PARAMS       0x04
 237 #define CONFIG_SUB_NODEMAP      0x08
 238 #define CONFIG_SUB_BARRIER      0x10
 239
 240 /* Sub clds should be attached to the config_llog_data when processing
 241  * config log for client or server target. */
 242 #define CONFIG_SUB_CLIENT       (CONFIG_SUB_SPTLRPC | CONFIG_SUB_RECOVER | \
 243                                  CONFIG_SUB_PARAMS)
 244 #define CONFIG_SUB_SERVER       (CONFIG_SUB_CLIENT | CONFIG_SUB_NODEMAP | \
 245                                  CONFIG_SUB_BARRIER)
 246
 247 #define PARAMS_FILENAME         "params"
 248 #define BARRIER_FILENAME        "barrier"
 249 #define LCTL_UPCALL             "lctl"
 250
 251 static inline bool logname_is_barrier(const char *logname)
 252 {
 253         char *ptr;
 254
 255         /* logname for barrier is "fsname-barrier" */
 256         ptr = strstr(logname, BARRIER_FILENAME);
 257         if (ptr && (ptr - logname) >= 2 &&
 258             *(ptr - 1) == '-' && *(ptr + 7) == '\0')
 259                 return true;
 260
 261         return false;
 262 }
 263
 264 /* list of active configuration logs  */
 265 struct config_llog_data {
 266         struct ldlm_res_id          cld_resid;
 267         struct lustre_handle        cld_lockh;
 268         struct config_llog_instance cld_cfg;
 269         struct list_head            cld_list_chain;/* on config_llog_list */
 270         atomic_t                    cld_refcount;
 271         struct config_llog_data    *cld_sptlrpc;/* depended sptlrpc log */
 272         struct config_llog_data    *cld_params; /* common parameters log */
 273         struct config_llog_data    *cld_recover;/* imperative recover log */
 274         struct config_llog_data    *cld_nodemap;/* nodemap log */
 275         struct config_llog_data    *cld_barrier;/* barrier log (for MDT only) */
 276         struct obd_export          *cld_mgcexp;
 277         struct mutex                cld_lock;
 278         enum mgs_cfg_type           cld_type;
 279         unsigned int                cld_stopping:1, /* we were told to stop
 280                                                      * watching */
 281                                     cld_lostlock:1; /* lock not requeued */
 282         char                        cld_logname[0];
 283 };
 284
 285 struct lustre_profile {
 286         struct list_head         lp_list;
 287         char                    *lp_profile;
 288         char                    *lp_dt;
 289         char                    *lp_md;
 290         int                      lp_refs;
 291         bool                     lp_list_deleted;
 292 };
 293
 294 struct lustre_profile *class_get_profile(const char * prof);
 295 void class_del_profile(const char *prof);
 296 void class_put_profile(struct lustre_profile *lprof);
 297 void class_del_profiles(void);
 298
 299
 300 #if LUSTRE_TRACKS_LOCK_EXP_REFS
 301
 302 void __class_export_add_lock_ref(struct obd_export *, struct ldlm_lock *);
 303 void __class_export_del_lock_ref(struct obd_export *, struct ldlm_lock *);
 304 extern void (*class_export_dump_hook)(struct obd_export *);
 305
 306 #else
 307
 308 #define __class_export_add_lock_ref(exp, lock)             do {} while (0)
 309 #define __class_export_del_lock_ref(exp, lock)             do {} while (0)
 310
 311 #endif
 312
 313 #define class_export_rpc_inc(exp)                                       \
 314 ({                                                                      \
 315         atomic_inc(&(exp)->exp_rpc_count);                              \
 316         CDEBUG(D_INFO, "RPC GETting export %p : new rpc_count %d\n",    \
 317                (exp), atomic_read(&(exp)->exp_rpc_count));              \
 318 })
 319
 320 #define class_export_rpc_dec(exp)                                       \
 321 ({                                                                      \
 322         LASSERT(atomic_read(&(exp)->exp_rpc_count) > 0);                \
 323         atomic_dec(&(exp)->exp_rpc_count);                              \
 324         CDEBUG(D_INFO, "RPC PUTting export %p : new rpc_count %d\n",    \
 325                (exp), atomic_read(&(exp)->exp_rpc_count));              \
 326 })
 327
 328 #define class_export_lock_get(exp, lock)                                \
 329 ({                                                                      \
 330         atomic_inc(&(exp)->exp_locks_count);                            \
 331         __class_export_add_lock_ref(exp, lock);                         \
 332         CDEBUG(D_INFO, "lock GETting export %p : new locks_count %d\n", \
 333                (exp), atomic_read(&(exp)->exp_locks_count));            \
 334         class_export_get(exp);                                          \
 335 })
 336
 337 #define class_export_lock_put(exp, lock)                                \
 338 ({                                                                      \
 339         LASSERT(atomic_read(&(exp)->exp_locks_count) > 0);              \
 340         atomic_dec(&(exp)->exp_locks_count);                            \
 341         __class_export_del_lock_ref(exp, lock);                         \
 342         CDEBUG(D_INFO, "lock PUTting export %p : new locks_count %d\n", \
 343                (exp), atomic_read(&(exp)->exp_locks_count));            \
 344         class_export_put(exp);                                          \
 345 })
 346
 347 #define class_export_cb_get(exp)                                        \
 348 ({                                                                      \
 349         atomic_inc(&(exp)->exp_cb_count);                               \
 350         CDEBUG(D_INFO, "callback GETting export %p : new cb_count %d\n",\
 351                (exp), atomic_read(&(exp)->exp_cb_count));               \
 352         class_export_get(exp);                                          \
 353 })
 354
 355 #define class_export_cb_put(exp)                                        \
 356 ({                                                                      \
 357         LASSERT(atomic_read(&(exp)->exp_cb_count) > 0);                 \
 358         atomic_dec(&(exp)->exp_cb_count);                               \
 359         CDEBUG(D_INFO, "callback PUTting export %p : new cb_count %d\n",\
 360                (exp), atomic_read(&(exp)->exp_cb_count));               \
 361         class_export_put(exp);                                          \
 362 })
 363
 364 /* genops.c */
 365 struct obd_export *class_export_get(struct obd_export *exp);
 366 void class_export_put(struct obd_export *exp);
 367 struct obd_export *class_new_export(struct obd_device *obd,
 368                                     struct obd_uuid *cluuid);
 369 struct obd_export *class_new_export_self(struct obd_device *obd,
 370                                          struct obd_uuid *uuid);
 371 void class_unlink_export(struct obd_export *exp);
 372
 373 struct obd_import *class_import_get(struct obd_import *);
 374 void class_import_put(struct obd_import *);
 375 struct obd_import *class_new_import(struct obd_device *obd);
 376 void class_destroy_import(struct obd_import *exp);
 377
 378 #ifdef HAVE_SERVER_SUPPORT
 379 struct obd_type *class_search_type(const char *name);
 380 struct obd_type *class_get_type(const char *name);
 381 #endif
 382 void class_put_type(struct obd_type *type);
 383 int class_connect(struct lustre_handle *conn, struct obd_device *obd,
 384                   struct obd_uuid *cluuid);
 385 int class_disconnect(struct obd_export *exp);
 386 void class_fail_export(struct obd_export *exp);
 387 int class_connected_export(struct obd_export *exp);
 388 void class_disconnect_exports(struct obd_device *obd);
 389 int class_manual_cleanup(struct obd_device *obd);
 390 void class_disconnect_stale_exports(struct obd_device *,
 391                                     int (*test_export)(struct obd_export *));
 392
 393 static inline enum obd_option exp_flags_from_obd(struct obd_device *obd)
 394 {
 395         return ((obd->obd_fail ? OBD_OPT_FAILOVER : 0) |
 396                 (obd->obd_force ? OBD_OPT_FORCE : 0) |
 397                 (obd->obd_abort_recovery ? OBD_OPT_ABORT_RECOV : 0) |
 398                 0);
 399 }
 400
 401 #ifdef HAVE_SERVER_SUPPORT
 402 static inline struct lu_target *class_exp2tgt(struct obd_export *exp)
 403 {
 404         struct obd_device_target *obt;
 405
 406         LASSERT(exp->exp_obd);
 407         obt = (void *)&exp->exp_obd->u;
 408         if (obt->obt_magic != OBT_MAGIC)
 409                 return NULL;
 410         return obt->obt_lut;
 411 }
 412
 413 static inline struct lr_server_data *class_server_data(struct obd_device *obd)
 414 {
 415         struct obd_device_target *obt = obd2obt(obd);
 416
 417         LASSERT(obt);
 418         return &obt->obt_lut->lut_lsd;
 419 }
 420 #endif
 421
 422 /* obdo.c */
 423 struct lu_attr;
 424 struct inode;
 425
 426 void obdo_from_la(struct obdo *dst, const struct lu_attr *la, u64 valid);
 427 void la_from_obdo(struct lu_attr *la, const struct obdo *dst, u64 valid);
 428
 429 void obdo_cpy_md(struct obdo *dst, const struct obdo *src, u64 valid);
 430 void obdo_to_ioobj(const struct obdo *oa, struct obd_ioobj *ioobj);
 431
 432 static inline int obd_check_dev(struct obd_device *obd)
 433 {
 434         if (!obd) {
 435                 CERROR("NULL device\n");
 436                 return -ENODEV;
 437         }
 438         return 0;
 439 }
 440
 441 /* ensure obd_setup and !obd_stopping */
 442 #define OBD_CHECK_DEV_ACTIVE(obd)                               \
 443 do {                                                            \
 444         rc = obd_check_dev(obd);                                \
 445         if (rc)                                                 \
 446                 return rc;                                      \
 447                                                                 \
 448         if (!(obd)->obd_set_up || (obd)->obd_stopping) {        \
 449                 CERROR("Device %d not setup\n",                 \
 450                        (obd)->obd_minor);                       \
 451                 RETURN(-ENODEV);                                \
 452         }                                                       \
 453 } while (0)
 454
 455
 456 static inline int lprocfs_nid_ldlm_stats_init(struct nid_stat* tmp)
 457 {
 458         /* Always add in ldlm_stats */
 459         tmp->nid_ldlm_stats =
 460                 lprocfs_stats_alloc(LDLM_LAST_OPC - LDLM_FIRST_OPC,
 461                                     LPROCFS_STATS_FLAG_NOPERCPU);
 462         if (tmp->nid_ldlm_stats == NULL)
 463                 return -ENOMEM;
 464
 465         lprocfs_init_ldlm_stats(tmp->nid_ldlm_stats);
 466
 467         return lprocfs_stats_register(tmp->nid_proc, "ldlm_stats",
 468                                       tmp->nid_ldlm_stats);
 469 }
 470
 471 static inline int exp_check_ops(struct obd_export *exp)
 472 {
 473         if (exp == NULL) {
 474                 RETURN(-ENODEV);
 475         }
 476         if (exp->exp_obd == NULL || !exp->exp_obd->obd_type) {
 477                 RETURN(-EOPNOTSUPP);
 478         }
 479         RETURN(0);
 480 }
 481
 482 static inline int obd_get_info(const struct lu_env *env, struct obd_export *exp,
 483                                __u32 keylen, void *key,
 484                                __u32 *vallen, void *val)
 485 {
 486         int rc;
 487
 488         ENTRY;
 489
 490         rc = exp_check_ops(exp);
 491         if (rc)
 492                 RETURN(rc);
 493
 494         if (!exp->exp_obd->obd_type->typ_dt_ops->o_get_info) {
 495                 CERROR("%s: no %s operation\n",
 496                        (exp)->exp_obd->obd_name, __func__);
 497                 RETURN(-ENOTSUPP);
 498         }
 499
 500         rc = exp->exp_obd->obd_type->typ_dt_ops->o_get_info(env, exp, keylen,
 501                                                             key, vallen, val);
 502         RETURN(rc);
 503 }
 504
 505 static inline int obd_set_info_async(const struct lu_env *env,
 506                                      struct obd_export *exp,
 507                                      __u32 keylen, void *key,
 508                                      __u32 vallen, void *val,
 509                                      struct ptlrpc_request_set *set)
 510 {
 511         int rc;
 512
 513         ENTRY;
 514
 515         rc = exp_check_ops(exp);
 516         if (rc)
 517                 RETURN(rc);
 518
 519         if (!exp->exp_obd->obd_type->typ_dt_ops->o_set_info_async) {
 520                 CERROR("%s: no %s operation\n",
 521                        (exp)->exp_obd->obd_name, __func__);
 522                 RETURN(-ENOTSUPP);
 523         }
 524
 525         rc = exp->exp_obd->obd_type->typ_dt_ops->o_set_info_async(env, exp,
 526                                                                   keylen,
 527                                                                   key,
 528                                                                   vallen,
 529                                                                   val, set);
 530         RETURN(rc);
 531 }
 532
 533 /*
 534  * obd-lu integration.
 535  *
 536  * Functionality is being moved into new lu_device-based layering, but some
 537  * pieces of configuration process are still based on obd devices.
 538  *
 539  * Specifically, lu_device_type_operations::ldto_device_alloc() methods fully
 540  * subsume ->o_setup() methods of obd devices they replace. The same for
 541  * lu_device_operations::ldo_process_config() and ->o_process_config(). As a
 542  * result, obd_setup() and obd_process_config() branch and call one XOR
 543  * another.
 544  *
 545  * Yet neither lu_device_type_operations::ldto_device_fini() nor
 546  * lu_device_type_operations::ldto_device_free() fully implement the
 547  * functionality of ->o_precleanup() and ->o_cleanup() they override. Hence,
 548  * obd_precleanup() and obd_cleanup() call both lu_device and obd operations.
 549  */
 550 static inline int obd_setup(struct obd_device *obd, struct lustre_cfg *cfg)
 551 {
 552         int rc;
 553         struct obd_type *type = obd->obd_type;
 554         struct lu_device_type *ldt;
 555
 556         ENTRY;
 557
 558         wait_var_event(&type->typ_lu,
 559                        smp_load_acquire(&type->typ_lu) != OBD_LU_TYPE_SETUP);
 560         ldt = type->typ_lu;
 561         if (ldt != NULL) {
 562                 struct lu_context session_ctx;
 563                 struct lu_env env;
 564
 565                 lu_context_init(&session_ctx, LCT_SESSION | LCT_SERVER_SESSION);
 566                 session_ctx.lc_thread = NULL;
 567                 lu_context_enter(&session_ctx);
 568
 569                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
 570                 if (rc == 0) {
 571                         struct lu_device *dev;
 572                         env.le_ses = &session_ctx;
 573                         dev = ldt->ldt_ops->ldto_device_alloc(&env, ldt, cfg);
 574                         lu_env_fini(&env);
 575                         if (!IS_ERR(dev)) {
 576                                 obd->obd_lu_dev = dev;
 577                                 dev->ld_obd = obd;
 578 #ifdef HAVE_SERVER_SUPPORT
 579                                 if (lu_device_is_dt(dev) &&
 580                                     lu2dt_dev(dev)->dd_rdonly)
 581                                         obd->obd_read_only = 1;
 582 #endif
 583                                 rc = 0;
 584                         } else
 585                                 rc = PTR_ERR(dev);
 586                 }
 587                 lu_context_exit(&session_ctx);
 588                 lu_context_fini(&session_ctx);
 589         } else {
 590                 if (!obd->obd_type->typ_dt_ops->o_setup) {
 591                         CERROR("%s: no %s operation\n", obd->obd_name,
 592                                __func__);
 593                         RETURN(-EOPNOTSUPP);
 594                 }
 595                 rc = obd->obd_type->typ_dt_ops->o_setup(obd, cfg);
 596         }
 597         RETURN(rc);
 598 }
 599
 600 static inline int obd_precleanup(struct obd_device *obd)
 601 {
 602         int rc;
 603         struct lu_device_type *ldt = obd->obd_type->typ_lu;
 604         struct lu_device *d = obd->obd_lu_dev;
 605
 606         ENTRY;
 607
 608         if (ldt != NULL && d != NULL) {
 609                 struct lu_env *env = lu_env_find();
 610                 struct lu_env _env;
 611
 612                 if (!env) {
 613                         env = &_env;
 614                         rc = lu_env_init(env, ldt->ldt_ctx_tags);
 615                         LASSERT(rc == 0);
 616                         lu_env_add(env);
 617                 }
 618                 ldt->ldt_ops->ldto_device_fini(env, d);
 619                 if (env == &_env) {
 620                         lu_env_remove(env);
 621                         lu_env_fini(env);
 622                 }
 623         }
 624
 625         if (!obd->obd_type->typ_dt_ops->o_precleanup)
 626                 RETURN(0);
 627
 628         rc = obd->obd_type->typ_dt_ops->o_precleanup(obd);
 629         RETURN(rc);
 630 }
 631
 632 static inline int obd_cleanup(struct obd_device *obd)
 633 {
 634         int rc;
 635         struct lu_device_type *ldt = obd->obd_type->typ_lu;
 636         struct lu_device *d = obd->obd_lu_dev;
 637
 638         ENTRY;
 639         if (ldt != NULL && d != NULL) {
 640                 struct lu_env env;
 641
 642                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
 643                 if (rc == 0) {
 644                         ldt->ldt_ops->ldto_device_free(&env, d);
 645                         lu_env_fini(&env);
 646                         obd->obd_lu_dev = NULL;
 647                 }
 648         }
 649         if (!obd->obd_type->typ_dt_ops->o_cleanup)
 650                 RETURN(0);
 651
 652         rc = obd->obd_type->typ_dt_ops->o_cleanup(obd);
 653         RETURN(rc);
 654 }
 655
 656 static inline void obd_cleanup_client_import(struct obd_device *obd)
 657 {
 658         ENTRY;
 659
 660         /* If we set up but never connected, the client import will not
 661          * have been cleaned.
 662          */
 663         down_write(&obd->u.cli.cl_sem);
 664         if (obd->u.cli.cl_import) {
 665                 struct obd_import *imp;
 666
 667                 imp = obd->u.cli.cl_import;
 668                 CDEBUG(D_CONFIG, "%s: client import never connected\n",
 669                        obd->obd_name);
 670                 ptlrpc_invalidate_import(imp);
 671                 client_destroy_import(imp);
 672                 obd->u.cli.cl_import = NULL;
 673         }
 674         up_write(&obd->u.cli.cl_sem);
 675
 676         EXIT;
 677 }
 678
 679 static inline int obd_process_config(struct obd_device *obd, int datalen,
 680                                      void *data)
 681 {
 682         int rc;
 683         struct lu_device_type *ldt = obd->obd_type->typ_lu;
 684         struct lu_device *d = obd->obd_lu_dev;
 685
 686         ENTRY;
 687
 688         obd->obd_process_conf = 1;
 689         if (ldt != NULL && d != NULL) {
 690                 struct lu_env env;
 691
 692                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
 693                 if (rc == 0) {
 694                         rc = d->ld_ops->ldo_process_config(&env, d, data);
 695                         lu_env_fini(&env);
 696                 }
 697         } else {
 698                 if (!obd->obd_type->typ_dt_ops->o_process_config) {
 699                         CERROR("%s: no %s operation\n",
 700                                obd->obd_name, __func__);
 701                         RETURN(-EOPNOTSUPP);
 702                 }
 703                 rc = obd->obd_type->typ_dt_ops->o_process_config(obd, datalen,
 704                                                                  data);
 705         }
 706
 707         obd->obd_process_conf = 0;
 708
 709         RETURN(rc);
 710 }
 711
 712 static inline int obd_create(const struct lu_env *env, struct obd_export *exp,
 713                              struct obdo *obdo)
 714 {
 715         int rc;
 716
 717         ENTRY;
 718
 719         rc = exp_check_ops(exp);
 720         if (rc)
 721                 RETURN(rc);
 722
 723         if (!exp->exp_obd->obd_type->typ_dt_ops->o_create) {
 724                 CERROR("%s: no %s operation\n",
 725                        (exp)->exp_obd->obd_name, __func__);
 726                 RETURN(-ENOTSUPP);
 727         }
 728
 729         rc = exp->exp_obd->obd_type->typ_dt_ops->o_create(env, exp, obdo);
 730         RETURN(rc);
 731 }
 732
 733 static inline int obd_destroy(const struct lu_env *env, struct obd_export *exp,
 734                               struct obdo *obdo)
 735 {
 736         int rc;
 737
 738         ENTRY;
 739
 740         rc = exp_check_ops(exp);
 741         if (rc)
 742                 RETURN(rc);
 743
 744         if (!exp->exp_obd->obd_type->typ_dt_ops->o_destroy) {
 745                 CERROR("%s: no %s operation\n",
 746                        (exp)->exp_obd->obd_name, __func__);
 747                 RETURN(-ENOTSUPP);
 748         }
 749
 750         rc = exp->exp_obd->obd_type->typ_dt_ops->o_destroy(env, exp, obdo);
 751         RETURN(rc);
 752 }
 753
 754 static inline int obd_getattr(const struct lu_env *env, struct obd_export *exp,
 755                               struct obdo *oa)
 756 {
 757         int rc;
 758
 759         ENTRY;
 760         rc = exp_check_ops(exp);
 761         if (rc)
 762                 RETURN(rc);
 763
 764         if (!exp->exp_obd->obd_type->typ_dt_ops->o_getattr) {
 765                 CERROR("%s: no %s operation\n",
 766                        (exp)->exp_obd->obd_name, __func__);
 767                 RETURN(-ENOTSUPP);
 768         }
 769
 770         rc = exp->exp_obd->obd_type->typ_dt_ops->o_getattr(env, exp, oa);
 771
 772         RETURN(rc);
 773 }
 774
 775 static inline int obd_setattr(const struct lu_env *env, struct obd_export *exp,
 776                               struct obdo *oa)
 777 {
 778         int rc;
 779
 780         ENTRY;
 781         rc = exp_check_ops(exp);
 782         if (rc)
 783                 RETURN(rc);
 784
 785         if (!exp->exp_obd->obd_type->typ_dt_ops->o_setattr) {
 786                 CERROR("%s: no %s operation\n",
 787                        (exp)->exp_obd->obd_name, __func__);
 788                 RETURN(-ENOTSUPP);
 789         }
 790
 791         rc = exp->exp_obd->obd_type->typ_dt_ops->o_setattr(env, exp, oa);
 792
 793         RETURN(rc);
 794 }
 795
 796 static inline int obd_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
 797                                int priority)
 798 {
 799         struct obd_device *obd = imp->imp_obd;
 800         int rc;
 801
 802         ENTRY;
 803
 804         OBD_CHECK_DEV_ACTIVE(obd);
 805         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_add_conn) {
 806                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
 807                 RETURN(-EOPNOTSUPP);
 808         }
 809
 810         rc = obd->obd_type->typ_dt_ops->o_add_conn(imp, uuid, priority);
 811         RETURN(rc);
 812 }
 813
 814 static inline int obd_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
 815 {
 816         struct obd_device *obd = imp->imp_obd;
 817         int rc;
 818
 819         ENTRY;
 820
 821         OBD_CHECK_DEV_ACTIVE(obd);
 822         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_del_conn) {
 823                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
 824                 RETURN(-EOPNOTSUPP);
 825         }
 826
 827         rc = obd->obd_type->typ_dt_ops->o_del_conn(imp, uuid);
 828         RETURN(rc);
 829 }
 830
 831 static inline struct obd_uuid *obd_get_uuid(struct obd_export *exp)
 832 {
 833         struct obd_uuid *uuid;
 834
 835         ENTRY;
 836
 837         if (!exp->exp_obd->obd_type ||
 838             !exp->exp_obd->obd_type->typ_dt_ops->o_get_uuid)
 839                 RETURN(NULL);
 840
 841         uuid = exp->exp_obd->obd_type->typ_dt_ops->o_get_uuid(exp);
 842         RETURN(uuid);
 843 }
 844
 845 /** Create a new /a exp on device /a obd for the uuid /a cluuid
 846  * @param exp New export handle
 847  * @param d Connect data, supported flags are set, flags also understood
 848  *    by obd are returned.
 849  */
 850 static inline int obd_connect(const struct lu_env *env,
 851                               struct obd_export **exp, struct obd_device *obd,
 852                               struct obd_uuid *cluuid,
 853                               struct obd_connect_data *data,
 854                               void *localdata)
 855 {
 856         int rc;
 857         __u64 ocf = data ? data->ocd_connect_flags : 0;
 858
 859         ENTRY;
 860
 861         OBD_CHECK_DEV_ACTIVE(obd);
 862         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_connect) {
 863                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
 864                 RETURN(-EOPNOTSUPP);
 865         }
 866
 867         rc = obd->obd_type->typ_dt_ops->o_connect(env, exp, obd, cluuid, data,
 868                                                   localdata);
 869         /* check that only subset is granted */
 870         LASSERT(ergo(data != NULL, (data->ocd_connect_flags & ocf) ==
 871                                     data->ocd_connect_flags));
 872         RETURN(rc);
 873 }
 874
 875 static inline int obd_reconnect(const struct lu_env *env,
 876                                 struct obd_export *exp,
 877                                 struct obd_device *obd,
 878                                 struct obd_uuid *cluuid,
 879                                 struct obd_connect_data *d,
 880                                 void *localdata)
 881 {
 882         int rc;
 883         __u64 ocf = d ? d->ocd_connect_flags : 0;
 884
 885         ENTRY;
 886
 887         OBD_CHECK_DEV_ACTIVE(obd);
 888         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_reconnect)
 889                 RETURN(0);
 890
 891         rc = obd->obd_type->typ_dt_ops->o_reconnect(env, exp, obd, cluuid, d,
 892                                                     localdata);
 893         /* check that only subset is granted */
 894         LASSERT(ergo(d != NULL,
 895                      (d->ocd_connect_flags & ocf) == d->ocd_connect_flags));
 896         RETURN(rc);
 897 }
 898
 899 static inline int obd_disconnect(struct obd_export *exp)
 900 {
 901         int rc;
 902
 903         ENTRY;
 904         rc = exp_check_ops(exp);
 905         if (rc)
 906                 RETURN(rc);
 907
 908         if (!exp->exp_obd->obd_type->typ_dt_ops->o_disconnect) {
 909                 CERROR("%s: no %s operation\n",
 910                        (exp)->exp_obd->obd_name, __func__);
 911                 RETURN(-ENOTSUPP);
 912         }
 913
 914         rc = exp->exp_obd->obd_type->typ_dt_ops->o_disconnect(exp);
 915         RETURN(rc);
 916 }
 917
 918 static inline int obd_fid_init(struct obd_device *obd, struct obd_export *exp,
 919                                enum lu_cli_type type)
 920 {
 921         int rc;
 922
 923         ENTRY;
 924
 925         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_fid_init)
 926                 RETURN(0);
 927
 928         rc = obd->obd_type->typ_dt_ops->o_fid_init(obd, exp, type);
 929         RETURN(rc);
 930 }
 931
 932 static inline int obd_fid_fini(struct obd_device *obd)
 933 {
 934         int rc;
 935
 936         ENTRY;
 937         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_fid_fini)
 938                 RETURN(0);
 939
 940         rc = obd->obd_type->typ_dt_ops->o_fid_fini(obd);
 941         RETURN(rc);
 942 }
 943
 944 static inline int obd_fid_alloc(const struct lu_env *env,
 945                                 struct obd_export *exp,
 946                                 struct lu_fid *fid,
 947                                 struct md_op_data *op_data)
 948 {
 949         int rc;
 950
 951         ENTRY;
 952         rc = exp_check_ops(exp);
 953         if (rc)
 954                 RETURN(rc);
 955
 956         if (!exp->exp_obd->obd_type->typ_dt_ops->o_fid_alloc) {
 957                 CERROR("%s: no %s operation\n",
 958                        (exp)->exp_obd->obd_name, __func__);
 959                 RETURN(-ENOTSUPP);
 960         }
 961
 962         rc = exp->exp_obd->obd_type->typ_dt_ops->o_fid_alloc(env, exp, fid,
 963                                                              op_data);
 964         RETURN(rc);
 965 }
 966
 967 static inline int obd_pool_new(struct obd_device *obd, char *poolname)
 968 {
 969         int rc;
 970
 971         ENTRY;
 972
 973         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_new) {
 974                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
 975                 RETURN(-EOPNOTSUPP);
 976         }
 977
 978         /* Check poolname validity */
 979         if (!poolname || poolname[0] == '\0' || lov_pool_is_reserved(poolname))
 980                 RETURN(-EINVAL);
 981
 982         rc = obd->obd_type->typ_dt_ops->o_pool_new(obd, poolname);
 983         RETURN(rc);
 984 }
 985
 986 static inline int obd_pool_del(struct obd_device *obd, char *poolname)
 987 {
 988         int rc;
 989
 990         ENTRY;
 991         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_del) {
 992                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
 993                 RETURN(-EOPNOTSUPP);
 994         }
 995
 996         rc = obd->obd_type->typ_dt_ops->o_pool_del(obd, poolname);
 997         RETURN(rc);
 998 }
 999
1000 static inline int obd_pool_add(struct obd_device *obd, char *poolname,
1001                                char *ostname)
1002 {
1003         int rc;
1004
1005         ENTRY;
1006
1007         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_add) {
1008                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
1009                 RETURN(-EOPNOTSUPP);
1010         }
1011
1012         rc = obd->obd_type->typ_dt_ops->o_pool_add(obd, poolname, ostname);
1013         RETURN(rc);
1014 }
1015
1016 static inline int obd_pool_rem(struct obd_device *obd, char *poolname,
1017                                char *ostname)
1018 {
1019         int rc;
1020
1021         ENTRY;
1022         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_rem) {
1023                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
1024                 RETURN(-EOPNOTSUPP);
1025         }
1026
1027         rc = obd->obd_type->typ_dt_ops->o_pool_rem(obd, poolname, ostname);
1028         RETURN(rc);
1029 }
1030
1031 static inline int obd_init_export(struct obd_export *exp)
1032 {
1033         int rc = 0;
1034
1035         ENTRY;
1036         if (exp->exp_obd != NULL && exp->exp_obd->obd_type &&
1037             exp->exp_obd->obd_type->typ_dt_ops->o_init_export)
1038                 rc = exp->exp_obd->obd_type->typ_dt_ops->o_init_export(exp);
1039         RETURN(rc);
1040 }
1041
1042 static inline int obd_destroy_export(struct obd_export *exp)
1043 {
1044         ENTRY;
1045         if (exp->exp_obd != NULL && exp->exp_obd->obd_type &&
1046             exp->exp_obd->obd_type->typ_dt_ops->o_destroy_export)
1047                 exp->exp_obd->obd_type->typ_dt_ops->o_destroy_export(exp);
1048         RETURN(0);
1049 }
1050
1051 /* @max_age is the oldest time in seconds that we accept using a cached data.
1052  * If the cache is older than @max_age we will get a new value from the
1053  * target. Use a value of 'ktime_get_seconds() + X' to guarantee freshness.
1054  */
1055 static inline int obd_statfs_async(struct obd_export *exp,
1056                                    struct obd_info *oinfo,
1057                                    time64_t max_age,
1058                                    struct ptlrpc_request_set *rqset)
1059 {
1060         struct obd_device *obd;
1061         int rc = 0;
1062
1063         ENTRY;
1064
1065         if (exp == NULL || exp->exp_obd == NULL)
1066                 RETURN(-EINVAL);
1067
1068         obd = exp->exp_obd;
1069         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_statfs) {
1070                 rc = -EOPNOTSUPP;
1071                 CERROR("%s: no statfs operation: rc = %d\n", obd->obd_name, rc);
1072                 RETURN(rc);
1073         }
1074
1075         CDEBUG(D_SUPER, "%s: age %lld, max_age %lld\n",
1076                obd->obd_name, obd->obd_osfs_age, max_age);
1077         rc = obd->obd_type->typ_dt_ops->o_statfs_async(exp, oinfo, max_age,
1078                                                        rqset);
1079
1080         RETURN(rc);
1081 }
1082
1083 /* @max_age is the oldest time in seconds that we accept using a cached data.
1084  * If the cache is older than @max_age we will get a new value from the
1085  * target. Use a value of 'ktime_get_seconds() + X' to guarantee freshness.
1086  */
1087 static inline int obd_statfs(const struct lu_env *env, struct obd_export *exp,
1088                              struct obd_statfs *osfs, time64_t max_age,
1089                              __u32 flags)
1090 {
1091         struct obd_device *obd;
1092         int rc = 0;
1093
1094         ENTRY;
1095         if (unlikely(exp == NULL || exp->exp_obd == NULL))
1096                 RETURN(-EINVAL);
1097
1098         obd = exp->exp_obd;
1099         OBD_CHECK_DEV_ACTIVE(obd);
1100
1101         if (unlikely(!obd->obd_type || !obd->obd_type->typ_dt_ops->o_statfs)) {
1102                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
1103                 RETURN(-EOPNOTSUPP);
1104         }
1105
1106         CDEBUG(D_SUPER, "%s: age %lld, max_age %lld\n",
1107                obd->obd_name, obd->obd_osfs_age, max_age);
1108         /* ignore cache if aggregated isn't expected */
1109         if (obd->obd_osfs_age < max_age ||
1110             ((obd->obd_osfs.os_state & OS_STATFS_SUM) &&
1111              !(flags & OBD_STATFS_SUM))) {
1112                 /* the RPC will block anyway, so avoid sending many at once */
1113                 rc = mutex_lock_interruptible_nested(&obd->obd_dev_mutex,
1114                                                      (flags & OBD_STATFS_NESTED)
1115                                                      ? SINGLE_DEPTH_NESTING : 0);
1116                 if (rc)
1117                         RETURN(rc);
1118                 if (obd->obd_osfs_age < max_age ||
1119                     ((obd->obd_osfs.os_state & OS_STATFS_SUM) &&
1120                      !(flags & OBD_STATFS_SUM))) {
1121                         rc = obd->obd_type->typ_dt_ops->o_statfs(env, exp, osfs,
1122                                                                  max_age,
1123                                                                  flags);
1124                 } else {
1125                         mutex_unlock(&obd->obd_dev_mutex);
1126                         GOTO(cached, rc = 0);
1127                 }
1128                 if (rc == 0) {
1129                         CDEBUG(D_SUPER,
1130                                "%s: update %p cache blocks %llu/%llu objects %llu/%llu\n",
1131                                obd->obd_name, &obd->obd_osfs,
1132                                osfs->os_bavail, osfs->os_blocks,
1133                                osfs->os_ffree, osfs->os_files);
1134
1135                         spin_lock(&obd->obd_osfs_lock);
1136                         memcpy(&obd->obd_osfs, osfs, sizeof(obd->obd_osfs));
1137                         obd->obd_osfs_age = ktime_get_seconds();
1138                         spin_unlock(&obd->obd_osfs_lock);
1139                 }
1140                 mutex_unlock(&obd->obd_dev_mutex);
1141         } else {
1142 cached:
1143                 CDEBUG(D_SUPER,
1144                        "%s: use %p cache blocks %llu/%llu objects %llu/%llu\n",
1145                        obd->obd_name, &obd->obd_osfs,
1146                        obd->obd_osfs.os_bavail, obd->obd_osfs.os_blocks,
1147                        obd->obd_osfs.os_ffree, obd->obd_osfs.os_files);
1148                 spin_lock(&obd->obd_osfs_lock);
1149                 memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
1150                 spin_unlock(&obd->obd_osfs_lock);
1151         }
1152         RETURN(rc);
1153 }
1154
1155 static inline int obd_preprw(const struct lu_env *env, int cmd,
1156                              struct obd_export *exp, struct obdo *oa,
1157                              int objcount, struct obd_ioobj *obj,
1158                              struct niobuf_remote *remote, int *pages,
1159                              struct niobuf_local *local)
1160 {
1161         int rc;
1162
1163         ENTRY;
1164
1165         rc = exp_check_ops(exp);
1166         if (rc)
1167                 RETURN(rc);
1168
1169         if (!exp->exp_obd->obd_type->typ_dt_ops->o_preprw) {
1170                 CERROR("%s: no %s operation\n",
1171                        (exp)->exp_obd->obd_name, __func__);
1172                 RETURN(-ENOTSUPP);
1173         }
1174
1175         rc = exp->exp_obd->obd_type->typ_dt_ops->o_preprw(env, cmd, exp,
1176                                                           oa, objcount, obj,
1177                                                           remote, pages, local);
1178
1179         RETURN(rc);
1180 }
1181
1182 static inline int obd_commitrw(const struct lu_env *env, int cmd,
1183                                struct obd_export *exp, struct obdo *oa,
1184                                int objcount, struct obd_ioobj *obj,
1185                                struct niobuf_remote *rnb, int pages,
1186                                struct niobuf_local *local, const int orig_rc,
1187                                int nob, ktime_t kstart)
1188 {
1189         int rc;
1190
1191         ENTRY;
1192
1193         rc = exp_check_ops(exp);
1194         if (rc)
1195                 RETURN(rc);
1196
1197         if (!exp->exp_obd->obd_type->typ_dt_ops->o_commitrw) {
1198                 CERROR("%s: no %s operation\n",
1199                        (exp)->exp_obd->obd_name, __func__);
1200                 RETURN(-ENOTSUPP);
1201         }
1202
1203         rc = exp->exp_obd->obd_type->typ_dt_ops->o_commitrw(env, cmd, exp, oa,
1204                                                             objcount, obj,
1205                                                             rnb, pages, local,
1206                                                             orig_rc, nob,
1207                                                             kstart);
1208
1209         RETURN(rc);
1210 }
1211
1212 static inline int obd_iocontrol(unsigned int cmd, struct obd_export *exp,
1213                                 int len, void *karg, void __user *uarg)
1214 {
1215         int rc;
1216
1217         ENTRY;
1218
1219         rc = exp_check_ops(exp);
1220         if (rc)
1221                 RETURN(rc);
1222
1223         if (!exp->exp_obd->obd_type->typ_dt_ops->o_iocontrol) {
1224                 CERROR("%s: no %s operation\n",
1225                        (exp)->exp_obd->obd_name, __func__);
1226                 RETURN(-ENOTSUPP);
1227         }
1228
1229         rc = exp->exp_obd->obd_type->typ_dt_ops->o_iocontrol(cmd, exp,
1230                                                              len, karg,
1231                                                              uarg);
1232         RETURN(rc);
1233 }
1234
1235 static inline void obd_import_event(struct obd_device *obd,
1236                                     struct obd_import *imp,
1237                                     enum obd_import_event event)
1238 {
1239         ENTRY;
1240         if (!obd) {
1241                 CERROR("NULL device\n");
1242                 EXIT;
1243                 return;
1244         }
1245
1246         if (obd->obd_set_up && obd->obd_type->typ_dt_ops->o_import_event)
1247                 obd->obd_type->typ_dt_ops->o_import_event(obd, imp, event);
1248
1249         EXIT;
1250 }
1251
1252 static inline int obd_notify(struct obd_device *obd,
1253                              struct obd_device *watched,
1254                              enum obd_notify_event ev)
1255 {
1256         int rc;
1257
1258         ENTRY;
1259
1260         rc = obd_check_dev(obd);
1261         if (rc)
1262                 return rc;
1263
1264         if (!obd->obd_set_up) {
1265                 CDEBUG(D_HA, "obd %s not set up\n", obd->obd_name);
1266                 RETURN(-EINVAL);
1267         }
1268
1269         if (!obd->obd_type->typ_dt_ops->o_notify) {
1270                 CDEBUG(D_HA, "obd %s has no notify handler\n", obd->obd_name);
1271                 RETURN(-ENOSYS);
1272         }
1273
1274         rc = obd->obd_type->typ_dt_ops->o_notify(obd, watched, ev);
1275
1276         RETURN(rc);
1277 }
1278
1279 static inline int obd_notify_observer(struct obd_device *observer,
1280                                       struct obd_device *observed,
1281                                       enum obd_notify_event ev)
1282 {
1283         int rc = 0;
1284         int rc2 = 0;
1285         struct obd_notify_upcall *onu;
1286
1287         if (WARN_ON_ONCE(!observer))
1288                 return -ENODEV;
1289
1290         if (observer->obd_observer)
1291                 rc = obd_notify(observer->obd_observer, observed, ev);
1292
1293         /*
1294          * Also, call non-obd listener, if any
1295          */
1296         onu = &observer->obd_upcall;
1297         if (onu->onu_upcall != NULL)
1298                 rc2 = onu->onu_upcall(observer, observed, ev, onu->onu_owner);
1299
1300         return rc ? rc : rc2;
1301 }
1302
1303 static inline int obd_quotactl(struct obd_export *exp,
1304                                struct obd_quotactl *oqctl)
1305 {
1306         int rc;
1307
1308         ENTRY;
1309
1310         rc = exp_check_ops(exp);
1311         if (rc)
1312                 RETURN(rc);
1313
1314         if (!exp->exp_obd->obd_type->typ_dt_ops->o_quotactl) {
1315                 CERROR("%s: no %s operation\n",
1316                        (exp)->exp_obd->obd_name, __func__);
1317                 RETURN(-ENOTSUPP);
1318         }
1319
1320         rc = exp->exp_obd->obd_type->typ_dt_ops->o_quotactl(exp->exp_obd,
1321                                                             exp, oqctl);
1322         RETURN(rc);
1323 }
1324
1325 static inline int obd_health_check(const struct lu_env *env,
1326                                    struct obd_device *obd)
1327 {
1328         /* returns: 0 on healthy
1329          *         >0 on unhealthy + reason code/flag
1330          *            however the only suppored reason == 1 right now
1331          *            We'll need to define some better reasons
1332          *            or flags in the future.
1333          *         <0 on error
1334          */
1335         int rc;
1336
1337         ENTRY;
1338
1339         /* NULL method is normal here */
1340         if (obd == NULL || !obd->obd_type) {
1341                 CERROR("cleaned up obd\n");
1342                 RETURN(-EOPNOTSUPP);
1343         }
1344         if (!obd->obd_set_up || obd->obd_stopping)
1345                 RETURN(0);
1346         if (!obd->obd_type->typ_dt_ops->o_health_check)
1347                 RETURN(0);
1348
1349         rc = obd->obd_type->typ_dt_ops->o_health_check(env, obd);
1350         RETURN(rc);
1351 }
1352
1353 static inline int obd_register_observer(struct obd_device *obd,
1354                                         struct obd_device *observer)
1355 {
1356         int rc;
1357
1358         ENTRY;
1359
1360         rc = obd_check_dev(obd);
1361         if (rc)
1362                 return rc;
1363
1364         down_write(&obd->obd_observer_link_sem);
1365         if (obd->obd_observer && observer) {
1366                 up_write(&obd->obd_observer_link_sem);
1367                 RETURN(-EALREADY);
1368         }
1369         obd->obd_observer = observer;
1370         up_write(&obd->obd_observer_link_sem);
1371         RETURN(0);
1372 }
1373
1374 /* metadata helpers */
1375 enum mps_stat_idx {
1376         LPROC_MD_CLOSE,
1377         LPROC_MD_CREATE,
1378         LPROC_MD_ENQUEUE,
1379         LPROC_MD_GETATTR,
1380         LPROC_MD_INTENT_LOCK,
1381         LPROC_MD_LINK,
1382         LPROC_MD_RENAME,
1383         LPROC_MD_SETATTR,
1384         LPROC_MD_FSYNC,
1385         LPROC_MD_READ_PAGE,
1386         LPROC_MD_UNLINK,
1387         LPROC_MD_SETXATTR,
1388         LPROC_MD_GETXATTR,
1389         LPROC_MD_INTENT_GETATTR_ASYNC,
1390         LPROC_MD_REVALIDATE_LOCK,
1391         LPROC_MD_LAST_OPC,
1392 };
1393
1394 static inline int md_get_root(struct obd_export *exp, const char *fileset,
1395                               struct lu_fid *fid)
1396 {
1397         int rc;
1398
1399         rc = exp_check_ops(exp);
1400         if (rc)
1401                 return rc;
1402
1403         return exp->exp_obd->obd_type->typ_md_ops->m_get_root(exp, fileset,
1404                                                               fid);
1405 }
1406
1407 static inline int md_getattr(struct obd_export *exp,
1408                              struct md_op_data *op_data,
1409                              struct ptlrpc_request **request)
1410 {
1411         int rc;
1412
1413         rc = exp_check_ops(exp);
1414         if (rc)
1415                 return rc;
1416
1417         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1418                              LPROC_MD_GETATTR);
1419
1420         return exp->exp_obd->obd_type->typ_md_ops->m_getattr(exp, op_data,
1421                                                              request);
1422 }
1423
1424 static inline int md_null_inode(struct obd_export *exp,
1425                                    const struct lu_fid *fid)
1426 {
1427         int rc;
1428
1429         rc = exp_check_ops(exp);
1430         if (rc)
1431                 return rc;
1432
1433         return exp->exp_obd->obd_type->typ_md_ops->m_null_inode(exp, fid);
1434 }
1435
1436 static inline int md_close(struct obd_export *exp, struct md_op_data *op_data,
1437                            struct md_open_data *mod,
1438                            struct ptlrpc_request **request)
1439 {
1440         int rc;
1441
1442         rc = exp_check_ops(exp);
1443         if (rc)
1444                 return rc;
1445
1446         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1447                              LPROC_MD_CLOSE);
1448
1449         return exp->exp_obd->obd_type->typ_md_ops->m_close(exp, op_data, mod,
1450                                                            request);
1451 }
1452
1453 static inline int md_create(struct obd_export *exp, struct md_op_data *op_data,
1454                             const void *data, size_t datalen, umode_t mode,
1455                             uid_t uid, gid_t gid, kernel_cap_t cap_effective,
1456                             __u64 rdev, struct ptlrpc_request **request)
1457 {
1458         int rc;
1459
1460         rc = exp_check_ops(exp);
1461         if (rc)
1462                 return rc;
1463
1464         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1465                              LPROC_MD_CREATE);
1466
1467         return exp->exp_obd->obd_type->typ_md_ops->m_create(exp, op_data, data,
1468                                                             datalen, mode, uid,
1469                                                             gid, cap_effective,
1470                                                             rdev, request);
1471 }
1472
1473 static inline int md_enqueue(struct obd_export *exp,
1474                              struct ldlm_enqueue_info *einfo,
1475                              const union ldlm_policy_data *policy,
1476                              struct md_op_data *op_data,
1477                              struct lustre_handle *lockh,
1478                              __u64 extra_lock_flags)
1479 {
1480         int rc;
1481
1482         rc = exp_check_ops(exp);
1483         if (rc)
1484                 return rc;
1485
1486         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1487                              LPROC_MD_ENQUEUE);
1488
1489         return exp->exp_obd->obd_type->typ_md_ops->m_enqueue(exp, einfo, policy,
1490                                                              op_data, lockh,
1491                                                              extra_lock_flags);
1492 }
1493
1494 static inline int md_getattr_name(struct obd_export *exp,
1495                                   struct md_op_data *op_data,
1496                                   struct ptlrpc_request **request)
1497 {
1498         int rc;
1499
1500         rc = exp_check_ops(exp);
1501         if (rc)
1502                 return rc;
1503
1504         return exp->exp_obd->obd_type->typ_md_ops->m_getattr_name(exp, op_data,
1505                                                                   request);
1506 }
1507
1508 static inline int md_intent_lock(struct obd_export *exp,
1509                                  struct md_op_data *op_data,
1510                                  struct lookup_intent *it,
1511                                  struct ptlrpc_request **reqp,
1512                                  ldlm_blocking_callback cb_blocking,
1513                                  __u64 extra_lock_flags)
1514 {
1515         int rc;
1516
1517         rc = exp_check_ops(exp);
1518         if (rc)
1519                 return rc;
1520
1521         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1522                              LPROC_MD_INTENT_LOCK);
1523
1524         return exp->exp_obd->obd_type->typ_md_ops->m_intent_lock(exp, op_data,
1525                                                                  it, reqp,
1526                                                                  cb_blocking,
1527                                                                  extra_lock_flags);
1528 }
1529
1530 static inline int md_link(struct obd_export *exp, struct md_op_data *op_data,
1531                           struct ptlrpc_request **request)
1532 {
1533         int rc;
1534
1535         rc = exp_check_ops(exp);
1536         if (rc)
1537                 return rc;
1538
1539         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1540                              LPROC_MD_LINK);
1541
1542         return exp->exp_obd->obd_type->typ_md_ops->m_link(exp, op_data,
1543                                                           request);
1544 }
1545
1546 static inline int md_rename(struct obd_export *exp, struct md_op_data *op_data,
1547                             const char *old_name, size_t oldlen,
1548                             const char *new_name, size_t newlen,
1549                             struct ptlrpc_request **request)
1550 {
1551         int rc;
1552
1553         rc = exp_check_ops(exp);
1554         if (rc)
1555                 return rc;
1556
1557         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1558                              LPROC_MD_RENAME);
1559
1560         return exp->exp_obd->obd_type->typ_md_ops->m_rename(exp, op_data,
1561                                                             old_name,
1562                                                             oldlen, new_name,
1563                                                             newlen, request);
1564 }
1565
1566 static inline int md_setattr(struct obd_export *exp, struct md_op_data *op_data,
1567                              void *ea, size_t ealen,
1568                              struct ptlrpc_request **request)
1569 {
1570         int rc;
1571
1572         rc = exp_check_ops(exp);
1573         if (rc)
1574                 return rc;
1575
1576         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1577                              LPROC_MD_SETATTR);
1578
1579         return exp->exp_obd->obd_type->typ_md_ops->m_setattr(exp, op_data, ea,
1580                                                              ealen, request);
1581 }
1582
1583 static inline int md_fsync(struct obd_export *exp, const struct lu_fid *fid,
1584                            struct ptlrpc_request **request)
1585 {
1586         int rc;
1587
1588         rc = exp_check_ops(exp);
1589         if (rc)
1590                 return rc;
1591
1592         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1593                              LPROC_MD_FSYNC);
1594
1595         return exp->exp_obd->obd_type->typ_md_ops->m_fsync(exp, fid, request);
1596 }
1597
1598 /* FLR: resync mirrored files. */
1599 static inline int md_file_resync(struct obd_export *exp,
1600                                  struct md_op_data *data)
1601 {
1602         int rc;
1603
1604         rc = exp_check_ops(exp);
1605         if (rc)
1606                 return rc;
1607
1608         return exp->exp_obd->obd_type->typ_md_ops->m_file_resync(exp, data);
1609 }
1610
1611 static inline int md_read_page(struct obd_export *exp,
1612                                struct md_op_data *op_data,
1613                                struct md_readdir_info *mrinfo,
1614                                __u64  hash_offset, struct page **ppage)
1615 {
1616         int rc;
1617
1618         rc = exp_check_ops(exp);
1619         if (rc)
1620                 return rc;
1621
1622         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1623                              LPROC_MD_READ_PAGE);
1624
1625         return exp->exp_obd->obd_type->typ_md_ops->m_read_page(exp, op_data,
1626                                                                mrinfo,
1627                                                                hash_offset,
1628                                                                ppage);
1629 }
1630
1631 static inline int md_unlink(struct obd_export *exp, struct md_op_data *op_data,
1632                             struct ptlrpc_request **request)
1633 {
1634         int rc;
1635
1636         rc = exp_check_ops(exp);
1637         if (rc)
1638                 return rc;
1639
1640         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1641                              LPROC_MD_UNLINK);
1642
1643         return exp->exp_obd->obd_type->typ_md_ops->m_unlink(exp, op_data,
1644                                                             request);
1645 }
1646
1647 static inline int md_get_lustre_md(struct obd_export *exp,
1648                                    struct req_capsule *pill,
1649                                    struct obd_export *dt_exp,
1650                                    struct obd_export *md_exp,
1651                                    struct lustre_md *md)
1652 {
1653         int rc;
1654
1655         rc = exp_check_ops(exp);
1656         if (rc)
1657                 return rc;
1658
1659         return exp->exp_obd->obd_type->typ_md_ops->m_get_lustre_md(exp, pill,
1660                                                                    dt_exp,
1661                                                                    md_exp, md);
1662 }
1663
1664 static inline int md_put_lustre_md(struct obd_export *exp,
1665                                     struct lustre_md *md)
1666 {
1667         int rc;
1668
1669         rc = exp_check_ops(exp);
1670         if (rc)
1671                 return rc;
1672
1673         return exp->exp_obd->obd_type->typ_md_ops->m_put_lustre_md(exp, md);
1674 }
1675
1676 static inline int md_merge_attr(struct obd_export *exp,
1677                                 const struct lmv_stripe_object *lso,
1678                                 struct cl_attr *attr,
1679                                 ldlm_blocking_callback cb)
1680 {
1681         int rc;
1682
1683         rc = exp_check_ops(exp);
1684         if (rc)
1685                 return rc;
1686
1687         return exp->exp_obd->obd_type->typ_md_ops->m_merge_attr(exp, lso, attr,
1688                                                                 cb);
1689 }
1690
1691 static inline int md_setxattr(struct obd_export *exp, const struct lu_fid *fid,
1692                               u64 obd_md_valid, const char *name,
1693                               const void *value, size_t value_size,
1694                               unsigned int xattr_flags, u32 suppgid,
1695                               struct ptlrpc_request **req)
1696 {
1697         int rc;
1698
1699         rc = exp_check_ops(exp);
1700         if (rc)
1701                 return rc;
1702
1703         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1704                              LPROC_MD_SETXATTR);
1705
1706         return exp->exp_obd->obd_type->typ_md_ops->m_setxattr(exp, fid,
1707                                                               obd_md_valid,
1708                                                               name, value,
1709                                                               value_size,
1710                                                               xattr_flags,
1711                                                               suppgid,
1712                                                               req);
1713 }
1714
1715 static inline int md_getxattr(struct obd_export *exp, const struct lu_fid *fid,
1716                               u64 obd_md_valid, const char *name,
1717                               size_t buf_size, struct ptlrpc_request **req)
1718 {
1719         int rc;
1720
1721         rc = exp_check_ops(exp);
1722         if (rc)
1723                 return rc;
1724
1725         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1726                              LPROC_MD_GETXATTR);
1727
1728         return exp->exp_obd->obd_type->typ_md_ops->m_getxattr(exp, fid,
1729                                                               obd_md_valid,
1730                                                               name, buf_size,
1731                                                               req);
1732 }
1733
1734 static inline int md_set_open_replay_data(struct obd_export *exp,
1735                                           struct obd_client_handle *och,
1736                                           struct lookup_intent *it)
1737 {
1738         int rc;
1739
1740         rc = exp_check_ops(exp);
1741         if (rc)
1742                 return rc;
1743
1744         return exp->exp_obd->obd_type->typ_md_ops->m_set_open_replay_data(exp,
1745                                                                           och,
1746                                                                           it);
1747 }
1748
1749 static inline int md_clear_open_replay_data(struct obd_export *exp,
1750                                             struct obd_client_handle *och)
1751 {
1752         int rc;
1753
1754         rc = exp_check_ops(exp);
1755         if (rc)
1756                 return rc;
1757
1758         return exp->exp_obd->obd_type->typ_md_ops->m_clear_open_replay_data(exp, och);
1759 }
1760
1761 static inline int md_set_lock_data(struct obd_export *exp,
1762                                    const struct lustre_handle *lockh,
1763                                    void *data, __u64 *bits)
1764 {
1765         int rc;
1766
1767         rc = exp_check_ops(exp);
1768         if (rc)
1769                 return rc;
1770
1771         return exp->exp_obd->obd_type->typ_md_ops->m_set_lock_data(exp, lockh,
1772                                                                    data,
1773                                                                    bits);
1774 }
1775
1776 static inline
1777 int md_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
1778                      union ldlm_policy_data *policy, enum ldlm_mode mode,
1779                      enum ldlm_cancel_flags cancel_flags, void *opaque)
1780 {
1781         int rc;
1782
1783         rc = exp_check_ops(exp);
1784         if (rc)
1785                 return rc;
1786
1787         return exp->exp_obd->obd_type->typ_md_ops->m_cancel_unused(exp,
1788                                                                    fid, policy,
1789                                                                    mode,
1790                                                                    cancel_flags,
1791                                                                    opaque);
1792 }
1793
1794 static inline enum ldlm_mode md_lock_match(struct obd_export *exp, __u64 flags,
1795                                            const struct lu_fid *fid,
1796                                            enum ldlm_type type,
1797                                            union ldlm_policy_data *policy,
1798                                            enum ldlm_mode mode,
1799                                            struct lustre_handle *lockh)
1800 {
1801         int rc;
1802
1803         rc = exp_check_ops(exp);
1804         if (rc)
1805                 return rc;
1806
1807         return exp->exp_obd->obd_type->typ_md_ops->m_lock_match(exp, flags,
1808                                                                 fid, type,
1809                                                                 policy, mode,
1810                                                                 lockh);
1811 }
1812
1813 static inline int md_init_ea_size(struct obd_export *exp, __u32 ea_size,
1814                                   __u32 def_ea_size)
1815 {
1816         int rc;
1817
1818         rc = exp_check_ops(exp);
1819         if (rc)
1820                 return rc;
1821
1822         return exp->exp_obd->obd_type->typ_md_ops->m_init_ea_size(exp, ea_size,
1823                                                                   def_ea_size);
1824 }
1825
1826 static inline int md_intent_getattr_async(struct obd_export *exp,
1827                                           struct md_op_item *item)
1828 {
1829         int rc;
1830
1831         rc = exp_check_ops(exp);
1832         if (rc)
1833                 return rc;
1834
1835         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1836                              LPROC_MD_INTENT_GETATTR_ASYNC);
1837
1838         return exp->exp_obd->obd_type->typ_md_ops->m_intent_getattr_async(exp,
1839                                                                           item);
1840 }
1841
1842 static inline int md_revalidate_lock(struct obd_export *exp,
1843                                      struct lookup_intent *it,
1844                                      struct lu_fid *fid, __u64 *bits)
1845 {
1846         int rc;
1847
1848         rc = exp_check_ops(exp);
1849         if (rc)
1850                 return rc;
1851
1852         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
1853                              LPROC_MD_REVALIDATE_LOCK);
1854
1855         return exp->exp_obd->obd_type->typ_md_ops->m_revalidate_lock(exp,
1856                                                                      it, fid,
1857                                                                      bits);
1858 }
1859
1860 static inline int md_get_fid_from_lsm(struct obd_export *exp,
1861                                       const struct lmv_stripe_object *lso,
1862                                       const char *name, int namelen,
1863                                       struct lu_fid *fid)
1864 {
1865         int rc;
1866
1867         rc = exp_check_ops(exp);
1868         if (rc)
1869                 return rc;
1870
1871         return exp->exp_obd->obd_type->typ_md_ops->m_get_fid_from_lsm(exp,
1872                                                                       lso, name,
1873                                                                       namelen,
1874                                                                       fid);
1875 }
1876
1877 /* Unpack an MD struct from disk to in-memory format.
1878  * Returns +ve size of unpacked MD (0 for free), or -ve error.
1879  *
1880  * If *plsm != NULL and lmm == NULL then *lsm will be freed.
1881  * If *plsm == NULL then it will be allocated.
1882  */
1883 static inline int md_stripe_object_create(struct obd_export *exp,
1884                                           struct lmv_stripe_object **lsop,
1885                                           const union lmv_mds_md *lmm,
1886                                           size_t lmm_size)
1887 {
1888         int rc;
1889
1890         rc = exp_check_ops(exp);
1891         if (rc)
1892                 return rc;
1893
1894         return exp->exp_obd->obd_type->typ_md_ops->m_stripe_object_create(exp,
1895                                                                           lsop,
1896                                                                           lmm,
1897                                                                           lmm_size);
1898 }
1899
1900 static inline int md_rmfid(struct obd_export *exp, struct fid_array *fa,
1901                            int *rcs, struct ptlrpc_request_set *set)
1902 {
1903         int rc;
1904
1905         rc = exp_check_ops(exp);
1906         if (rc)
1907                 return rc;
1908
1909         return exp->exp_obd->obd_type->typ_md_ops->m_rmfid(exp, fa, rcs, set);
1910 }
1911
1912 static inline struct lu_batch *
1913 md_batch_create(struct obd_export *exp, enum lu_batch_flags flags,
1914                 __u32 max_count)
1915 {
1916         int rc;
1917
1918         rc = exp_check_ops(exp);
1919         if (rc)
1920                 return ERR_PTR(rc);
1921
1922         return exp->exp_obd->obd_type->typ_md_ops->m_batch_create(exp, flags,
1923                                                                   max_count);
1924 }
1925
1926 static inline int md_batch_stop(struct obd_export *exp, struct lu_batch *bh)
1927 {
1928         int rc;
1929
1930         rc = exp_check_ops(exp);
1931         if (rc)
1932                 return rc;
1933
1934         return exp->exp_obd->obd_type->typ_md_ops->m_batch_stop(exp, bh);
1935 }
1936
1937 static inline int md_batch_flush(struct obd_export *exp, struct lu_batch *bh,
1938                                  bool wait)
1939 {
1940         int rc;
1941
1942         rc = exp_check_ops(exp);
1943         if (rc)
1944                 return rc;
1945
1946         return exp->exp_obd->obd_type->typ_md_ops->m_batch_flush(exp, bh, wait);
1947 }
1948
1949 static inline int md_batch_add(struct obd_export *exp, struct lu_batch *bh,
1950                                struct md_op_item *item)
1951 {
1952         int rc;
1953
1954         rc = exp_check_ops(exp);
1955         if (rc)
1956                 return rc;
1957
1958         return exp->exp_obd->obd_type->typ_md_ops->m_batch_add(exp, bh, item);
1959 }
1960
1961 /* OBD Metadata Support */
1962
1963 extern int obd_init_caches(void);
1964 extern void obd_cleanup_caches(void);
1965
1966 typedef int (*register_lwp_cb)(void *data);
1967
1968 struct lwp_register_item {
1969         struct obd_export **lri_exp;
1970         register_lwp_cb     lri_cb_func;
1971         void               *lri_cb_data;
1972         struct list_head    lri_list;
1973         atomic_t            lri_ref;
1974         char                lri_name[MTI_NAME_MAXLEN];
1975 };
1976
1977 /* obd_mount.c */
1978 #ifdef HAVE_SERVER_SUPPORT
1979 int lustre_register_lwp_item(const char *lwpname, struct obd_export **exp,
1980                              register_lwp_cb cb_func, void *cb_data);
1981 void lustre_deregister_lwp_item(struct obd_export **exp);
1982 struct obd_export *lustre_find_lwp_by_index(const char *dev, __u32 idx);
1983 void lustre_notify_lwp_list(struct obd_export *exp);
1984 int tgt_name2lwp_name(const char *tgt_name, char *lwp_name, int len, __u32 idx);
1985 #endif /* HAVE_SERVER_SUPPORT */
1986 int lustre_check_exclusion(struct super_block *sb, char *svname);
1987
1988 /* lustre_peer.c    */
1989 int lustre_uuid_to_peer(const char *uuid, struct lnet_nid *peer_nid,
1990                         int index);
1991 int class_add_uuid(const char *uuid, struct lnet_nid *nid);
1992 int class_del_uuid (const char *uuid);
1993 int class_add_nids_to_uuid(struct obd_uuid *uuid, struct lnet_nid *nidlist,
1994                            int nid_count, int nid_size);
1995 int class_check_uuid(struct obd_uuid *uuid, struct lnet_nid *nid);
1996
1997 /* class_obd.c */
1998 extern char obd_jobid_name[];
1999
2000 extern unsigned int obd_lbug_on_eviction;
2001 extern unsigned int obd_dump_on_eviction;
2002
2003 static inline bool do_dump_on_eviction(struct obd_device *exp_obd)
2004 {
2005         if (obd_lbug_on_eviction &&
2006             strncmp(exp_obd->obd_type->typ_name, LUSTRE_MGC_NAME,
2007                     strlen(LUSTRE_MGC_NAME))) {
2008                 CERROR("LBUG upon eviction\n");
2009                 LBUG();
2010         }
2011
2012         return obd_dump_on_eviction;
2013 }
2014
2015 /* statfs_pack.c */
2016 struct kstatfs;
2017 void statfs_pack(struct obd_statfs *osfs, struct kstatfs *sfs);
2018 void statfs_unpack(struct kstatfs *sfs, struct obd_statfs *osfs);
2019
2020 /* root squash info */
2021 struct root_squash_info {
2022         uid_t                   rsi_uid;
2023         gid_t                   rsi_gid;
2024         struct list_head        rsi_nosquash_nids;
2025         spinlock_t              rsi_lock;
2026 };
2027
2028 int server_name2index(const char *svname, __u32 *idx, const char **endptr);
2029
2030 /* linux-module.c */
2031 struct obd_ioctl_data;
2032 int obd_ioctl_getdata(struct obd_ioctl_data **data, int *len, void __user *arg);
2033 int class_procfs_init(void);
2034 int class_procfs_clean(void);
2035
2036 extern void obd_heat_add(struct obd_heat_instance *instance,
2037                          unsigned int time_second, __u64 count,
2038                          unsigned int weight, unsigned int period_second);
2039 extern void obd_heat_decay(struct obd_heat_instance *instance,
2040                            __u64 time_second, unsigned int weight,
2041                            unsigned int period_second);
2042 extern __u64 obd_heat_get(struct obd_heat_instance *instance,
2043                           unsigned int time_second, unsigned int weight,
2044                           unsigned int period_second);
2045 extern void obd_heat_clear(struct obd_heat_instance *instance, int count);
2046
2047 /* struct kobj_type */
2048 static inline
2049 struct attribute *_get_attr_matches(const struct kobj_type *typ,
2050                                     const char *key, size_t keylen,
2051                                     int (*is_match)(const char *, const char *,
2052                                                     size_t))
2053 {
2054         int i;
2055
2056 #ifdef HAVE_KOBJ_TYPE_DEFAULT_GROUPS
2057         for (i = 0; typ->default_groups[i]; i++) {
2058                 int k;
2059                 struct attribute **attrs;
2060
2061                 attrs = (struct attribute **)typ->default_groups[i]->attrs;
2062                 for (k = 0; attrs[k]; k++) {
2063                         if (is_match(attrs[k]->name, key, keylen))
2064                                 return (struct attribute *)attrs[k];
2065                 }
2066         }
2067 #else
2068         for (i = 0; typ->default_attrs[i]; i++) {
2069                 if (is_match(typ->default_attrs[i]->name, key, keylen))
2070                         return typ->default_attrs[i];
2071         }
2072 #endif
2073         return NULL;
2074 }
2075
2076 static inline
2077 int _attr_name_exact(const char *attr_name, const char *key, size_t len)
2078 {
2079         return !strcmp(attr_name, key);
2080 }
2081
2082 static inline
2083 struct attribute *get_attr_by_name(const struct kobj_type *typ,
2084                                    const char *name)
2085 {
2086         return _get_attr_matches(typ, name, 0, _attr_name_exact);
2087 }
2088
2089 static inline
2090 int _attr_name_starts_with(const char *attr_name, const char *name, size_t len)
2091 {
2092         return !strncmp(attr_name, name, len);
2093 }
2094
2095 static inline
2096 struct attribute *get_attr_starts_with(const struct kobj_type *typ,
2097                                        const char *name,
2098                                        size_t len)
2099 {
2100         return _get_attr_matches(typ, name, len, _attr_name_starts_with);
2101 }
2102
2103 int obd_ioctl_msg(const char *file, const char *func, int line, int level,
2104                   const char *name, unsigned int cmd, const char *msg, int rc);
2105 #define OBD_IOC_DEBUG(level, dev, cmd, msg, rc) \
2106         obd_ioctl_msg(__FILE__, __func__, __LINE__, level, dev, cmd, msg, rc)
2107 #define OBD_IOC_ERROR(dev, cmd, msg, rc)        \
2108         obd_ioctl_msg(__FILE__, __func__, __LINE__, D_ERROR, dev, cmd, msg, rc)
2109
2110 #endif /* __LINUX_OBD_CLASS_H */