Whamcloud - gitweb
LU-11052 obd: remove OBD ops based stats
[fs/lustre-release.git] / lustre / include / obd_class.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  */
32 #ifndef __CLASS_OBD_H
33 #define __CLASS_OBD_H
34
35 #include <linux/kobject.h>
36 #include <obd_support.h>
37 #include <lustre_import.h>
38 #include <lustre_net.h>
39 #include <obd.h>
40 #include <lustre_lib.h>
41 #include <uapi/linux/lustre/lustre_idl.h>
42 #include <lprocfs_status.h>
43
44 #define OBD_STATFS_NODELAY      0x0001  /* requests should be send without delay
45                                          * and resends for avoid deadlocks */
46 #define OBD_STATFS_FROM_CACHE   0x0002  /* the statfs callback should not update
47                                          * obd_osfs_age */
48 #define OBD_STATFS_FOR_MDT0     0x0004  /* The statfs is only for retrieving
49                                          * information from MDT0. */
50
51 extern rwlock_t obd_dev_lock;
52
53 /* OBD Operations Declarations */
54 extern struct obd_device *class_conn2obd(struct lustre_handle *);
55 extern struct obd_device *class_exp2obd(struct obd_export *);
56 extern int class_handle_ioctl(unsigned int cmd, unsigned long arg);
57 int lustre_get_jobid(char *jobid, size_t len);
58 void lustre_jobid_clear(const char *jobid);
59 void jobid_cache_fini(void);
60 int jobid_cache_init(void);
61
62 struct lu_device_type;
63
64 /* genops.c */
65 struct obd_export *class_conn2export(struct lustre_handle *);
66 struct kobject *class_setup_tunables(const char *name);
67 int class_register_type(struct obd_ops *, struct md_ops *, bool enable_proc,
68                         struct lprocfs_vars *module_vars,
69                         const char *nm, struct lu_device_type *ldt);
70 int class_unregister_type(const char *nm);
71
72 struct obd_device *class_newdev(const char *type_name, const char *name,
73                                 const char *uuid);
74 int class_register_device(struct obd_device *obd);
75 void class_unregister_device(struct obd_device *obd);
76 void class_free_dev(struct obd_device *obd);
77
78 struct obd_device *class_dev_by_str(const char *str);
79 int class_name2dev(const char *name);
80 struct obd_device *class_name2obd(const char *name);
81 int class_uuid2dev(struct obd_uuid *uuid);
82 struct obd_device *class_uuid2obd(struct obd_uuid *uuid);
83 void class_obd_list(void);
84 struct obd_device * class_find_client_obd(struct obd_uuid *tgt_uuid,
85                                           const char * typ_name,
86                                           struct obd_uuid *grp_uuid);
87 struct obd_device * class_devices_in_group(struct obd_uuid *grp_uuid,
88                                            int *next);
89 struct obd_device * class_num2obd(int num);
90 int get_devices_count(void);
91
92 int class_notify_sptlrpc_conf(const char *fsname, int namelen);
93
94 static inline char *obd_export_nid2str(struct obd_export *exp)
95 {
96         return exp->exp_connection == NULL ?
97                "<unknown>" : libcfs_nid2str(exp->exp_connection->c_peer.nid);
98 }
99
100 static inline char *obd_import_nid2str(struct obd_import *imp)
101 {
102         return imp->imp_connection == NULL ?
103                "<unknown>" : libcfs_nid2str(imp->imp_connection->c_peer.nid);
104 }
105
106 int obd_export_evict_by_nid(struct obd_device *obd, const char *nid);
107 int obd_export_evict_by_uuid(struct obd_device *obd, const char *uuid);
108 int obd_connect_flags2str(char *page, int count, __u64 flags, __u64 flags2,
109                           const char *sep);
110
111 int obd_zombie_impexp_init(void);
112 void obd_zombie_impexp_stop(void);
113 void obd_zombie_impexp_cull(void);
114 void obd_zombie_barrier(void);
115 void obd_exports_barrier(struct obd_device *obd);
116 int kuc_len(int payload_len);
117 struct kuc_hdr * kuc_ptr(void *p);
118 void *kuc_alloc(int payload_len, int transport, int type);
119 void kuc_free(void *p, int payload_len);
120 int obd_get_request_slot(struct client_obd *cli);
121 void obd_put_request_slot(struct client_obd *cli);
122 __u32 obd_get_max_rpcs_in_flight(struct client_obd *cli);
123 int obd_set_max_rpcs_in_flight(struct client_obd *cli, __u32 max);
124 __u16 obd_get_max_mod_rpcs_in_flight(struct client_obd *cli);
125 int obd_set_max_mod_rpcs_in_flight(struct client_obd *cli, __u16 max);
126 int obd_mod_rpc_stats_seq_show(struct client_obd *cli, struct seq_file *seq);
127
128 __u16 obd_get_mod_rpc_slot(struct client_obd *cli, __u32 opc,
129                            struct lookup_intent *it);
130 void obd_put_mod_rpc_slot(struct client_obd *cli, __u32 opc,
131                           struct lookup_intent *it, __u16 tag);
132
133 struct llog_handle;
134 struct llog_rec_hdr;
135 typedef int (*llog_cb_t)(const struct lu_env *, struct llog_handle *,
136                          struct llog_rec_hdr *, void *);
137
138 struct obd_export *obd_stale_export_get(void);
139 void obd_stale_export_put(struct obd_export *exp);
140 void obd_stale_export_adjust(struct obd_export *exp);
141
142 /* obd_config.c */
143 /* For interoperability */
144 struct cfg_interop_param {
145         char *old_param;
146         char *new_param;
147 };
148
149 char *lustre_cfg_string(struct lustre_cfg *lcfg, u32 index);
150 struct lustre_cfg *lustre_cfg_rename(struct lustre_cfg *cfg,
151                                      const char *new_name);
152 void print_lustre_cfg(struct lustre_cfg *lcfg);
153 int class_process_config(struct lustre_cfg *lcfg);
154 ssize_t class_set_global(const char *param);
155 ssize_t class_modify_config(struct lustre_cfg *lcfg, const char *prefix,
156                             struct kobject *kobj);
157 int class_process_proc_param(char *prefix, struct lprocfs_vars *lvars,
158                              struct lustre_cfg *lcfg, void *data);
159 int class_attach(struct lustre_cfg *lcfg);
160 int class_setup(struct obd_device *obd, struct lustre_cfg *lcfg);
161 int class_cleanup(struct obd_device *obd, struct lustre_cfg *lcfg);
162 int class_detach(struct obd_device *obd, struct lustre_cfg *lcfg);
163
164 int class_find_param(char *buf, char *key, char **valp);
165 struct cfg_interop_param *class_find_old_param(const char *param,
166                                                struct cfg_interop_param *ptr);
167 int class_get_next_param(char **params, char *copy);
168 int class_match_param(char *buf, const char *key, char **valp);
169 int class_parse_nid(char *buf, lnet_nid_t *nid, char **endh);
170 int class_parse_nid_quiet(char *buf, lnet_nid_t *nid, char **endh);
171 int class_parse_net(char *buf, u32 *net, char **endh);
172 int class_match_nid(char *buf, char *key, lnet_nid_t nid);
173 int class_match_net(char *buf, char *key, u32 net);
174
175 struct obd_device *class_incref(struct obd_device *obd,
176                                 const char *scope, const void *source);
177 void class_decref(struct obd_device *obd,
178                   const char *scope, const void *source);
179 void dump_exports(struct obd_device *obd, int locks, int debug_level);
180 int class_config_llog_handler(const struct lu_env *env,
181                               struct llog_handle *handle,
182                               struct llog_rec_hdr *rec, void *data);
183 int class_add_conn(struct obd_device *obd, struct lustre_cfg *lcfg);
184
185 #define CFG_F_START     0x01   /* Set when we start updating from a log */
186 #define CFG_F_MARKER    0x02   /* We are within a maker */
187 #define CFG_F_SKIP      0x04   /* We should ignore this cfg command */
188 #define CFG_F_EXCLUDE   0x10   /* OST exclusion list */
189
190 /* Passed as data param to class_config_parse_llog */
191 struct config_llog_instance {
192         char                    *cfg_obdname;
193         void                    *cfg_instance;
194         struct super_block      *cfg_sb;
195         struct obd_uuid          cfg_uuid;
196         llog_cb_t                cfg_callback;
197         int                      cfg_last_idx; /* for partial llog processing */
198         int                      cfg_flags;
199         __u32                    cfg_lwp_idx;
200         __u32                    cfg_sub_clds;
201 };
202 int class_config_parse_llog(const struct lu_env *env, struct llog_ctxt *ctxt,
203                             char *name, struct config_llog_instance *cfg);
204
205 #define CONFIG_SUB_SPTLRPC      0x01
206 #define CONFIG_SUB_RECOVER      0x02
207 #define CONFIG_SUB_PARAMS       0x04
208 #define CONFIG_SUB_NODEMAP      0x08
209 #define CONFIG_SUB_BARRIER      0x10
210
211 /* Sub clds should be attached to the config_llog_data when processing
212  * config log for client or server target. */
213 #define CONFIG_SUB_CLIENT       (CONFIG_SUB_SPTLRPC | CONFIG_SUB_RECOVER | \
214                                  CONFIG_SUB_PARAMS)
215 #define CONFIG_SUB_SERVER       (CONFIG_SUB_CLIENT | CONFIG_SUB_NODEMAP | \
216                                  CONFIG_SUB_BARRIER)
217
218 #define PARAMS_FILENAME         "params"
219 #define BARRIER_FILENAME        "barrier"
220 #define LCTL_UPCALL             "lctl"
221
222 static inline bool logname_is_barrier(const char *logname)
223 {
224         char *ptr;
225
226         /* logname for barrier is "fsname-barrier" */
227         ptr = strstr(logname, BARRIER_FILENAME);
228         if (ptr && (ptr - logname) >= 2 &&
229             *(ptr - 1) == '-' && *(ptr + 7) == '\0')
230                 return true;
231
232         return false;
233 }
234
235 /* list of active configuration logs  */
236 struct config_llog_data {
237         struct ldlm_res_id          cld_resid;
238         struct config_llog_instance cld_cfg;
239         struct list_head            cld_list_chain;/* on config_llog_list */
240         atomic_t                    cld_refcount;
241         struct config_llog_data    *cld_sptlrpc;/* depended sptlrpc log */
242         struct config_llog_data    *cld_params; /* common parameters log */
243         struct config_llog_data    *cld_recover;/* imperative recover log */
244         struct config_llog_data    *cld_nodemap;/* nodemap log */
245         struct config_llog_data    *cld_barrier;/* barrier log (for MDT only) */
246         struct obd_export          *cld_mgcexp;
247         struct mutex                cld_lock;
248         int                         cld_type;
249         unsigned int                cld_stopping:1, /* we were told to stop
250                                                      * watching */
251                                     cld_lostlock:1; /* lock not requeued */
252         char                        cld_logname[0];
253 };
254
255 struct lustre_profile {
256         struct list_head         lp_list;
257         char                    *lp_profile;
258         char                    *lp_dt;
259         char                    *lp_md;
260         int                      lp_refs;
261         bool                     lp_list_deleted;
262 };
263
264 struct lustre_profile *class_get_profile(const char * prof);
265 void class_del_profile(const char *prof);
266 void class_put_profile(struct lustre_profile *lprof);
267 void class_del_profiles(void);
268
269
270 #if LUSTRE_TRACKS_LOCK_EXP_REFS
271
272 void __class_export_add_lock_ref(struct obd_export *, struct ldlm_lock *);
273 void __class_export_del_lock_ref(struct obd_export *, struct ldlm_lock *);
274 extern void (*class_export_dump_hook)(struct obd_export *);
275
276 #else
277
278 #define __class_export_add_lock_ref(exp, lock)             do {} while(0)
279 #define __class_export_del_lock_ref(exp, lock)             do {} while(0)
280
281 #endif
282
283 #define class_export_rpc_inc(exp)                                       \
284 ({                                                                      \
285         atomic_inc(&(exp)->exp_rpc_count);                              \
286         CDEBUG(D_INFO, "RPC GETting export %p : new rpc_count %d\n",    \
287                (exp), atomic_read(&(exp)->exp_rpc_count));              \
288 })
289
290 #define class_export_rpc_dec(exp)                                       \
291 ({                                                                      \
292         LASSERT_ATOMIC_POS(&exp->exp_rpc_count);                        \
293         atomic_dec(&(exp)->exp_rpc_count);                              \
294         CDEBUG(D_INFO, "RPC PUTting export %p : new rpc_count %d\n",    \
295                (exp), atomic_read(&(exp)->exp_rpc_count));              \
296 })
297
298 #define class_export_lock_get(exp, lock)                                \
299 ({                                                                      \
300         atomic_inc(&(exp)->exp_locks_count);                            \
301         __class_export_add_lock_ref(exp, lock);                         \
302         CDEBUG(D_INFO, "lock GETting export %p : new locks_count %d\n", \
303                (exp), atomic_read(&(exp)->exp_locks_count));            \
304         class_export_get(exp);                                          \
305 })
306
307 #define class_export_lock_put(exp, lock)                                \
308 ({                                                                      \
309         LASSERT_ATOMIC_POS(&exp->exp_locks_count);                      \
310         atomic_dec(&(exp)->exp_locks_count);                            \
311         __class_export_del_lock_ref(exp, lock);                         \
312         CDEBUG(D_INFO, "lock PUTting export %p : new locks_count %d\n", \
313                (exp), atomic_read(&(exp)->exp_locks_count));            \
314         class_export_put(exp);                                          \
315 })
316
317 #define class_export_cb_get(exp)                                        \
318 ({                                                                      \
319         atomic_inc(&(exp)->exp_cb_count);                               \
320         CDEBUG(D_INFO, "callback GETting export %p : new cb_count %d\n",\
321                (exp), atomic_read(&(exp)->exp_cb_count));               \
322         class_export_get(exp);                                          \
323 })
324
325 #define class_export_cb_put(exp)                                        \
326 ({                                                                      \
327         LASSERT_ATOMIC_POS(&exp->exp_cb_count);                         \
328         atomic_dec(&(exp)->exp_cb_count);                               \
329         CDEBUG(D_INFO, "callback PUTting export %p : new cb_count %d\n",\
330                (exp), atomic_read(&(exp)->exp_cb_count));               \
331         class_export_put(exp);                                          \
332 })
333
334 /* genops.c */
335 struct obd_export *class_export_get(struct obd_export *exp);
336 void class_export_put(struct obd_export *exp);
337 struct obd_export *class_new_export(struct obd_device *obddev,
338                                     struct obd_uuid *cluuid);
339 struct obd_export *class_new_export_self(struct obd_device *obd,
340                                          struct obd_uuid *uuid);
341 void class_unlink_export(struct obd_export *exp);
342
343 struct obd_import *class_import_get(struct obd_import *);
344 void class_import_put(struct obd_import *);
345 struct obd_import *class_new_import(struct obd_device *obd);
346 void class_destroy_import(struct obd_import *exp);
347
348 struct obd_type *class_search_type(const char *name);
349 struct obd_type *class_get_type(const char *name);
350 void class_put_type(struct obd_type *type);
351 int class_connect(struct lustre_handle *conn, struct obd_device *obd,
352                   struct obd_uuid *cluuid);
353 int class_disconnect(struct obd_export *exp);
354 void class_fail_export(struct obd_export *exp);
355 int class_connected_export(struct obd_export *exp);
356 void class_disconnect_exports(struct obd_device *obddev);
357 int class_manual_cleanup(struct obd_device *obd);
358 void class_disconnect_stale_exports(struct obd_device *,
359                                     int (*test_export)(struct obd_export *));
360 static inline enum obd_option exp_flags_from_obd(struct obd_device *obd)
361 {
362         return ((obd->obd_fail ? OBD_OPT_FAILOVER : 0) |
363                 (obd->obd_force ? OBD_OPT_FORCE : 0) |
364                 (obd->obd_abort_recovery ? OBD_OPT_ABORT_RECOV : 0) |
365                 0);
366 }
367
368 #ifdef HAVE_SERVER_SUPPORT
369 static inline struct lu_target *class_exp2tgt(struct obd_export *exp)
370 {
371         LASSERT(exp->exp_obd);
372         if (exp->exp_obd->u.obt.obt_magic != OBT_MAGIC)
373                 return NULL;
374         return exp->exp_obd->u.obt.obt_lut;
375 }
376
377 static inline struct lr_server_data *class_server_data(struct obd_device *obd)
378 {
379         LASSERT(obd->u.obt.obt_lut);
380         return &obd->u.obt.obt_lut->lut_lsd;
381 }
382 #endif
383
384 /* obdo.c */
385 struct lu_attr;
386 struct inode;
387
388 void obdo_from_la(struct obdo *dst, const struct lu_attr *la, u64 valid);
389 void la_from_obdo(struct lu_attr *la, const struct obdo *dst, u64 valid);
390
391 void obdo_cpy_md(struct obdo *dst, const struct obdo *src, u64 valid);
392 void obdo_to_ioobj(const struct obdo *oa, struct obd_ioobj *ioobj);
393
394 #define OBP(dev, op)    (dev)->obd_type->typ_dt_ops->o_ ## op
395 #define MDP(dev, op)    (dev)->obd_type->typ_md_ops->m_ ## op
396
397 /* Ensure obd_setup: used for cleanup which must be called
398    while obd is stopping */
399 #define OBD_CHECK_DEV(obd)                                      \
400 do {                                                            \
401         if (!(obd)) {                                           \
402                 CERROR("NULL device\n");                        \
403                 RETURN(-ENODEV);                                \
404         }                                                       \
405 } while (0)
406
407 /* ensure obd_setup and !obd_stopping */
408 #define OBD_CHECK_DEV_ACTIVE(obd)                               \
409 do {                                                            \
410         OBD_CHECK_DEV(obd);                                     \
411         if (!(obd)->obd_set_up || (obd)->obd_stopping) {        \
412                 CERROR("Device %d not setup\n",                 \
413                        (obd)->obd_minor);                       \
414                 RETURN(-ENODEV);                                \
415         }                                                       \
416 } while (0)
417
418
419 #ifdef CONFIG_PROC_FS
420 #define _MD_COUNTER_OFFSET(m_op)                                               \
421         ((offsetof(struct md_ops, m_op) -                                      \
422           offsetof(struct md_ops, MD_STATS_FIRST_OP)) /                        \
423          sizeof(((struct md_ops *)NULL)->MD_STATS_FIRST_OP))
424
425 #define MD_COUNTER_OFFSET(op) _MD_COUNTER_OFFSET(m_ ## op)
426
427 #define NUM_MD_STATS                                                           \
428         (_MD_COUNTER_OFFSET(MD_STATS_LAST_OP) -                                \
429          _MD_COUNTER_OFFSET(MD_STATS_FIRST_OP) + 1)
430
431 /* Note that we only increment md counters for ops whose offset is less
432  * than NUM_MD_STATS. This is explained in a comment in the definition
433  * of struct md_ops. */
434 #define EXP_MD_COUNTER_INCREMENT(exp, op)                                      \
435         do {                                                                   \
436                 if (MD_COUNTER_OFFSET(op) < NUM_MD_STATS)                      \
437                         lprocfs_counter_incr((exp)->exp_obd->obd_md_stats,     \
438                                         (exp)->exp_obd->obd_md_cntr_base +     \
439                                         MD_COUNTER_OFFSET(op));                \
440         } while (0)
441
442 #else
443 #define EXP_MD_COUNTER_INCREMENT(exp, op)
444 #endif
445
446 static inline int lprocfs_nid_ldlm_stats_init(struct nid_stat* tmp)
447 {
448         /* Always add in ldlm_stats */
449         tmp->nid_ldlm_stats =
450                 lprocfs_alloc_stats(LDLM_LAST_OPC - LDLM_FIRST_OPC,
451                                     LPROCFS_STATS_FLAG_NOPERCPU);
452         if (tmp->nid_ldlm_stats == NULL)
453                 return -ENOMEM;
454
455         lprocfs_init_ldlm_stats(tmp->nid_ldlm_stats);
456
457         return lprocfs_register_stats(tmp->nid_proc, "ldlm_stats",
458                                       tmp->nid_ldlm_stats);
459 }
460
461 static inline int exp_check_ops(struct obd_export *exp)
462 {
463         if (exp == NULL) {
464                 RETURN(-ENODEV);
465         }
466         if (exp->exp_obd == NULL || !exp->exp_obd->obd_type) {
467                 RETURN(-EOPNOTSUPP);
468         }
469         RETURN(0);
470 }
471
472 static inline int class_devno_max(void)
473 {
474         return MAX_OBD_DEVICES;
475 }
476
477 static inline int obd_get_info(const struct lu_env *env, struct obd_export *exp,
478                                __u32 keylen, void *key,
479                                __u32 *vallen, void *val)
480 {
481         int rc;
482         ENTRY;
483
484         rc = exp_check_ops(exp);
485         if (rc)
486                 RETURN(rc);
487
488         if (!exp->exp_obd->obd_type->typ_dt_ops->o_get_info) {
489                 CERROR("%s: no %s operation\n",
490                        (exp)->exp_obd->obd_name, __func__);
491                 RETURN(-ENOTSUPP);
492         }
493
494         rc = OBP(exp->exp_obd, get_info)(env, exp, keylen, key, vallen, val);
495         RETURN(rc);
496 }
497
498 static inline int obd_set_info_async(const struct lu_env *env,
499                                      struct obd_export *exp,
500                                      __u32 keylen, void *key,
501                                      __u32 vallen, void *val,
502                                      struct ptlrpc_request_set *set)
503 {
504         int rc;
505         ENTRY;
506
507         rc = exp_check_ops(exp);
508         if (rc)
509                 RETURN(rc);
510
511         if (!exp->exp_obd->obd_type->typ_dt_ops->o_set_info_async) {
512                 CERROR("%s: no %s operation\n",
513                        (exp)->exp_obd->obd_name, __func__);
514                 RETURN(-ENOTSUPP);
515         }
516
517         rc = OBP(exp->exp_obd, set_info_async)(env, exp, keylen, key, vallen,
518                                                val, set);
519         RETURN(rc);
520 }
521
522 /*
523  * obd-lu integration.
524  *
525  * Functionality is being moved into new lu_device-based layering, but some
526  * pieces of configuration process are still based on obd devices.
527  *
528  * Specifically, lu_device_type_operations::ldto_device_alloc() methods fully
529  * subsume ->o_setup() methods of obd devices they replace. The same for
530  * lu_device_operations::ldo_process_config() and ->o_process_config(). As a
531  * result, obd_setup() and obd_process_config() branch and call one XOR
532  * another.
533  *
534  * Yet neither lu_device_type_operations::ldto_device_fini() nor
535  * lu_device_type_operations::ldto_device_free() fully implement the
536  * functionality of ->o_precleanup() and ->o_cleanup() they override. Hence,
537  * obd_precleanup() and obd_cleanup() call both lu_device and obd operations.
538  */
539 static inline int obd_setup(struct obd_device *obd, struct lustre_cfg *cfg)
540 {
541         int rc;
542         struct lu_device_type *ldt = obd->obd_type->typ_lu;
543         struct lu_device *d;
544
545         ENTRY;
546
547         if (ldt != NULL) {
548                 struct lu_context  session_ctx;
549                 struct lu_env env;
550                 lu_context_init(&session_ctx, LCT_SESSION | LCT_SERVER_SESSION);
551                 session_ctx.lc_thread = NULL;
552                 lu_context_enter(&session_ctx);
553
554                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
555                 if (rc == 0) {
556                         env.le_ses = &session_ctx;
557                         d = ldt->ldt_ops->ldto_device_alloc(&env, ldt, cfg);
558                         lu_env_fini(&env);
559                         if (!IS_ERR(d)) {
560                                 obd->obd_lu_dev = d;
561                                 d->ld_obd = obd;
562                                 rc = 0;
563                         } else
564                                 rc = PTR_ERR(d);
565                 }
566                 lu_context_exit(&session_ctx);
567                 lu_context_fini(&session_ctx);
568
569         } else {
570                 if (!obd->obd_type->typ_dt_ops->o_setup) {
571                         CERROR("%s: no %s operation\n", obd->obd_name,
572                                __func__);
573                         RETURN(-EOPNOTSUPP);
574                 }
575                 rc = OBP(obd, setup)(obd, cfg);
576         }
577         RETURN(rc);
578 }
579
580 static inline int obd_precleanup(struct obd_device *obd)
581 {
582         int rc;
583         struct lu_device_type *ldt = obd->obd_type->typ_lu;
584         struct lu_device *d = obd->obd_lu_dev;
585
586         ENTRY;
587
588         if (ldt != NULL && d != NULL) {
589                 struct lu_env env;
590
591                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
592                 if (rc == 0) {
593                         ldt->ldt_ops->ldto_device_fini(&env, d);
594                         lu_env_fini(&env);
595                 }
596         }
597
598         if (!obd->obd_type->typ_dt_ops->o_precleanup)
599                 RETURN(0);
600
601         rc = OBP(obd, precleanup)(obd);
602         RETURN(rc);
603 }
604
605 static inline int obd_cleanup(struct obd_device *obd)
606 {
607         int rc;
608         struct lu_device_type *ldt = obd->obd_type->typ_lu;
609         struct lu_device *d = obd->obd_lu_dev;
610
611         ENTRY;
612         if (ldt != NULL && d != NULL) {
613                 struct lu_env env;
614
615                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
616                 if (rc == 0) {
617                         ldt->ldt_ops->ldto_device_free(&env, d);
618                         lu_env_fini(&env);
619                         obd->obd_lu_dev = NULL;
620                 }
621         }
622         if (!obd->obd_type->typ_dt_ops->o_cleanup)
623                 RETURN(0);
624
625         rc = OBP(obd, cleanup)(obd);
626         RETURN(rc);
627 }
628
629 static inline void obd_cleanup_client_import(struct obd_device *obd)
630 {
631         ENTRY;
632
633         /* If we set up but never connected, the
634            client import will not have been cleaned. */
635         down_write(&obd->u.cli.cl_sem);
636         if (obd->u.cli.cl_import) {
637                 struct obd_import *imp;
638                 imp = obd->u.cli.cl_import;
639                 CDEBUG(D_CONFIG, "%s: client import never connected\n",
640                        obd->obd_name);
641                 ptlrpc_invalidate_import(imp);
642                 client_destroy_import(imp);
643                 obd->u.cli.cl_import = NULL;
644         }
645         up_write(&obd->u.cli.cl_sem);
646
647         EXIT;
648 }
649
650 static inline int obd_process_config(struct obd_device *obd, int datalen,
651                                      void *data)
652 {
653         int rc;
654         struct lu_device_type *ldt = obd->obd_type->typ_lu;
655         struct lu_device *d = obd->obd_lu_dev;
656
657         ENTRY;
658
659         obd->obd_process_conf = 1;
660         if (ldt != NULL && d != NULL) {
661                 struct lu_env env;
662
663                 rc = lu_env_init(&env, ldt->ldt_ctx_tags);
664                 if (rc == 0) {
665                         rc = d->ld_ops->ldo_process_config(&env, d, data);
666                         lu_env_fini(&env);
667                 }
668         } else {
669                 if (!obd->obd_type->typ_dt_ops->o_process_config) {
670                         CERROR("%s: no %s operation\n",
671                                obd->obd_name, __func__);
672                         RETURN(-EOPNOTSUPP);
673                 }
674                 rc = OBP(obd, process_config)(obd, datalen, data);
675         }
676
677         obd->obd_process_conf = 0;
678
679         RETURN(rc);
680 }
681
682 static inline int obd_create(const struct lu_env *env, struct obd_export *exp,
683                              struct obdo *obdo)
684 {
685         int rc;
686         ENTRY;
687
688         rc = exp_check_ops(exp);
689         if (rc)
690                 RETURN(rc);
691
692         if (!exp->exp_obd->obd_type->typ_dt_ops->o_create) {
693                 CERROR("%s: no %s operation\n",
694                        (exp)->exp_obd->obd_name, __func__);
695                 RETURN(-ENOTSUPP);
696         }
697
698         rc = OBP(exp->exp_obd, create)(env, exp, obdo);
699         RETURN(rc);
700 }
701
702 static inline int obd_destroy(const struct lu_env *env, struct obd_export *exp,
703                               struct obdo *obdo)
704 {
705         int rc;
706         ENTRY;
707
708         rc = exp_check_ops(exp);
709         if (rc)
710                 RETURN(rc);
711
712         if (!exp->exp_obd->obd_type->typ_dt_ops->o_destroy) {
713                 CERROR("%s: no %s operation\n",
714                        (exp)->exp_obd->obd_name, __func__);
715                 RETURN(-ENOTSUPP);
716         }
717
718         rc = OBP(exp->exp_obd, destroy)(env, exp, obdo);
719         RETURN(rc);
720 }
721
722 static inline int obd_getattr(const struct lu_env *env, struct obd_export *exp,
723                               struct obdo *oa)
724 {
725         int rc;
726
727         ENTRY;
728         rc = exp_check_ops(exp);
729         if (rc)
730                 RETURN(rc);
731
732         if (!exp->exp_obd->obd_type->typ_dt_ops->o_getattr) {
733                 CERROR("%s: no %s operation\n",
734                        (exp)->exp_obd->obd_name, __func__);
735                 RETURN(-ENOTSUPP);
736         }
737
738         rc = OBP(exp->exp_obd, getattr)(env, exp, oa);
739
740         RETURN(rc);
741 }
742
743 static inline int obd_setattr(const struct lu_env *env, struct obd_export *exp,
744                               struct obdo *oa)
745 {
746         int rc;
747
748         ENTRY;
749         rc = exp_check_ops(exp);
750         if (rc)
751                 RETURN(rc);
752
753         if (!exp->exp_obd->obd_type->typ_dt_ops->o_setattr) {
754                 CERROR("%s: no %s operation\n",
755                        (exp)->exp_obd->obd_name, __func__);
756                 RETURN(-ENOTSUPP);
757         }
758
759         rc = OBP(exp->exp_obd, setattr)(env, exp, oa);
760
761         RETURN(rc);
762 }
763
764 static inline int obd_add_conn(struct obd_import *imp, struct obd_uuid *uuid,
765                                int priority)
766 {
767         struct obd_device *obd = imp->imp_obd;
768         int rc;
769         ENTRY;
770
771         OBD_CHECK_DEV_ACTIVE(obd);
772         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_add_conn) {
773                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
774                 RETURN(-EOPNOTSUPP);
775         }
776
777         rc = OBP(obd, add_conn)(imp, uuid, priority);
778         RETURN(rc);
779 }
780
781 static inline int obd_del_conn(struct obd_import *imp, struct obd_uuid *uuid)
782 {
783         struct obd_device *obd = imp->imp_obd;
784         int rc;
785         ENTRY;
786
787         OBD_CHECK_DEV_ACTIVE(obd);
788         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_del_conn) {
789                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
790                 RETURN(-EOPNOTSUPP);
791         }
792
793         rc = OBP(obd, del_conn)(imp, uuid);
794         RETURN(rc);
795 }
796
797 static inline struct obd_uuid *obd_get_uuid(struct obd_export *exp)
798 {
799         struct obd_uuid *uuid;
800         ENTRY;
801
802         if (!exp->exp_obd->obd_type ||
803             !exp->exp_obd->obd_type->typ_dt_ops->o_get_uuid)
804                 RETURN(NULL);
805
806         uuid = OBP(exp->exp_obd, get_uuid)(exp);
807         RETURN(uuid);
808 }
809
810 /** Create a new /a exp on device /a obd for the uuid /a cluuid
811  * @param exp New export handle
812  * @param d Connect data, supported flags are set, flags also understood
813  *    by obd are returned.
814  */
815 static inline int obd_connect(const struct lu_env *env,
816                               struct obd_export **exp,struct obd_device *obd,
817                               struct obd_uuid *cluuid,
818                               struct obd_connect_data *data,
819                               void *localdata)
820 {
821         int rc;
822         __u64 ocf = data ? data->ocd_connect_flags : 0; /* for post-condition
823                                                    * check */
824         ENTRY;
825
826         OBD_CHECK_DEV_ACTIVE(obd);
827         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_connect) {
828                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
829                 RETURN(-EOPNOTSUPP);
830         }
831
832         rc = OBP(obd, connect)(env, exp, obd, cluuid, data, localdata);
833         /* check that only subset is granted */
834         LASSERT(ergo(data != NULL, (data->ocd_connect_flags & ocf) ==
835                                     data->ocd_connect_flags));
836         RETURN(rc);
837 }
838
839 static inline int obd_reconnect(const struct lu_env *env,
840                                 struct obd_export *exp,
841                                 struct obd_device *obd,
842                                 struct obd_uuid *cluuid,
843                                 struct obd_connect_data *d,
844                                 void *localdata)
845 {
846         int rc;
847         __u64 ocf = d ? d->ocd_connect_flags : 0; /* for post-condition
848                                                    * check */
849
850         ENTRY;
851
852         OBD_CHECK_DEV_ACTIVE(obd);
853         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_reconnect)
854                 RETURN(0);
855
856         rc = OBP(obd, reconnect)(env, exp, obd, cluuid, d, localdata);
857         /* check that only subset is granted */
858         LASSERT(ergo(d != NULL,
859                      (d->ocd_connect_flags & ocf) == d->ocd_connect_flags));
860         RETURN(rc);
861 }
862
863 static inline int obd_disconnect(struct obd_export *exp)
864 {
865         int rc;
866         ENTRY;
867         rc = exp_check_ops(exp);
868         if (rc)
869                 RETURN(rc);
870
871         if (!exp->exp_obd->obd_type->typ_dt_ops->o_disconnect) {
872                 CERROR("%s: no %s operation\n",
873                        (exp)->exp_obd->obd_name, __func__);
874                 RETURN(-ENOTSUPP);
875         }
876
877         rc = OBP(exp->exp_obd, disconnect)(exp);
878         RETURN(rc);
879 }
880
881 static inline int obd_fid_init(struct obd_device *obd, struct obd_export *exp,
882                                enum lu_cli_type type)
883 {
884         int rc;
885         ENTRY;
886
887         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_fid_init)
888                 RETURN(0);
889
890         rc = OBP(obd, fid_init)(obd, exp, type);
891         RETURN(rc);
892 }
893
894 static inline int obd_fid_fini(struct obd_device *obd)
895 {
896         int rc;
897         ENTRY;
898         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_fid_fini)
899                 RETURN(0);
900
901         rc = OBP(obd, fid_fini)(obd);
902         RETURN(rc);
903 }
904
905 static inline int obd_fid_alloc(const struct lu_env *env,
906                                 struct obd_export *exp,
907                                 struct lu_fid *fid,
908                                 struct md_op_data *op_data)
909 {
910         int rc;
911         ENTRY;
912         rc = exp_check_ops(exp);
913         if (rc)
914                 RETURN(rc);
915
916         if (!exp->exp_obd->obd_type->typ_dt_ops->o_fid_alloc) {
917                 CERROR("%s: no %s operation\n",
918                        (exp)->exp_obd->obd_name, __func__);
919                 RETURN(-ENOTSUPP);
920         }
921
922         rc = OBP(exp->exp_obd, fid_alloc)(env, exp, fid, op_data);
923         RETURN(rc);
924 }
925
926 static inline int obd_ping(const struct lu_env *env, struct obd_export *exp)
927 {
928         int rc;
929         ENTRY;
930
931         if (!exp->exp_obd->obd_type ||
932             !exp->exp_obd->obd_type->typ_dt_ops->o_ping)
933                 RETURN(0);
934
935         rc = OBP(exp->exp_obd, ping)(env, exp);
936         RETURN(rc);
937 }
938
939 static inline int obd_pool_new(struct obd_device *obd, char *poolname)
940 {
941         int rc;
942         ENTRY;
943
944         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_new) {
945                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
946                 RETURN(-EOPNOTSUPP);
947         }
948
949         rc = OBP(obd, pool_new)(obd, poolname);
950         RETURN(rc);
951 }
952
953 static inline int obd_pool_del(struct obd_device *obd, char *poolname)
954 {
955         int rc;
956         ENTRY;
957         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_del) {
958                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
959                 RETURN(-EOPNOTSUPP);
960         }
961
962         rc = OBP(obd, pool_del)(obd, poolname);
963         RETURN(rc);
964 }
965
966 static inline int obd_pool_add(struct obd_device *obd, char *poolname,
967                                char *ostname)
968 {
969         int rc;
970         ENTRY;
971
972         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_add) {
973                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
974                 RETURN(-EOPNOTSUPP);
975         }
976
977         rc = OBP(obd, pool_add)(obd, poolname, ostname);
978         RETURN(rc);
979 }
980
981 static inline int obd_pool_rem(struct obd_device *obd, char *poolname,
982                                char *ostname)
983 {
984         int rc;
985
986         ENTRY;
987         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_pool_rem) {
988                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
989                 RETURN(-EOPNOTSUPP);
990         }
991
992         rc = OBP(obd, pool_rem)(obd, poolname, ostname);
993         RETURN(rc);
994 }
995
996 static inline int obd_init_export(struct obd_export *exp)
997 {
998         int rc = 0;
999
1000         ENTRY;
1001         if (exp->exp_obd != NULL && exp->exp_obd->obd_type &&
1002             OBP((exp)->exp_obd, init_export))
1003                 rc = OBP(exp->exp_obd, init_export)(exp);
1004         RETURN(rc);
1005 }
1006
1007 static inline int obd_destroy_export(struct obd_export *exp)
1008 {
1009         ENTRY;
1010         if (exp->exp_obd != NULL && exp->exp_obd->obd_type &&
1011             OBP(exp->exp_obd, destroy_export))
1012                 OBP(exp->exp_obd, destroy_export)(exp);
1013         RETURN(0);
1014 }
1015
1016 /* @max_age is the oldest time in seconds that we accept using a cached data.
1017  * If the cache is older than @max_age we will get a new value from the
1018  * target. Use a value of 'ktime_get_seconds() + X' to guarantee freshness.
1019  */
1020 static inline int obd_statfs_async(struct obd_export *exp,
1021                                    struct obd_info *oinfo,
1022                                    time64_t max_age,
1023                                    struct ptlrpc_request_set *rqset)
1024 {
1025         int rc = 0;
1026         struct obd_device *obd;
1027
1028         ENTRY;
1029
1030         if (exp == NULL || exp->exp_obd == NULL)
1031                 RETURN(-EINVAL);
1032
1033         obd = exp->exp_obd;
1034         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_statfs) {
1035                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
1036                 RETURN(-EOPNOTSUPP);
1037         }
1038
1039         CDEBUG(D_SUPER, "%s: osfs %p age %lld, max_age %lld\n",
1040                obd->obd_name, &obd->obd_osfs, obd->obd_osfs_age, max_age);
1041         if (obd->obd_osfs_age < max_age) {
1042                 rc = OBP(obd, statfs_async)(exp, oinfo, max_age, rqset);
1043         } else {
1044                 CDEBUG(D_SUPER,
1045                        "%s: use %p cache blocks %llu/%llu objects %llu/%llu\n",
1046                        obd->obd_name, &obd->obd_osfs,
1047                        obd->obd_osfs.os_bavail, obd->obd_osfs.os_blocks,
1048                        obd->obd_osfs.os_ffree, obd->obd_osfs.os_files);
1049                 spin_lock(&obd->obd_osfs_lock);
1050                 memcpy(oinfo->oi_osfs, &obd->obd_osfs,
1051                        sizeof(*oinfo->oi_osfs));
1052                 spin_unlock(&obd->obd_osfs_lock);
1053                 oinfo->oi_flags |= OBD_STATFS_FROM_CACHE;
1054                 if (oinfo->oi_cb_up)
1055                         oinfo->oi_cb_up(oinfo, 0);
1056         }
1057         RETURN(rc);
1058 }
1059
1060 /* @max_age is the oldest time in seconds that we accept using a cached data.
1061  * If the cache is older than @max_age we will get a new value from the
1062  * target. Use a value of 'ktime_get_seconds() + X' to guarantee freshness.
1063  */
1064 static inline int obd_statfs(const struct lu_env *env, struct obd_export *exp,
1065                              struct obd_statfs *osfs, time64_t max_age,
1066                              __u32 flags)
1067 {
1068         int rc = 0;
1069         struct obd_device *obd = exp->exp_obd;
1070         ENTRY;
1071
1072         if (obd == NULL)
1073                 RETURN(-EINVAL);
1074
1075         OBD_CHECK_DEV_ACTIVE(obd);
1076
1077         if (!obd->obd_type || !obd->obd_type->typ_dt_ops->o_statfs) {
1078                 CERROR("%s: no %s operation\n", obd->obd_name, __func__);
1079                 RETURN(-EOPNOTSUPP);
1080         }
1081
1082         CDEBUG(D_SUPER, "osfs %lld, max_age %lld\n",
1083                obd->obd_osfs_age, max_age);
1084         if (obd->obd_osfs_age < max_age) {
1085                 rc = OBP(obd, statfs)(env, exp, osfs, max_age, flags);
1086                 if (rc == 0) {
1087                         spin_lock(&obd->obd_osfs_lock);
1088                         memcpy(&obd->obd_osfs, osfs, sizeof(obd->obd_osfs));
1089                         obd->obd_osfs_age = ktime_get_seconds();
1090                         spin_unlock(&obd->obd_osfs_lock);
1091                 }
1092         } else {
1093                 CDEBUG(D_SUPER, "%s: use %p cache blocks %llu/%llu"
1094                        " objects %llu/%llu\n",
1095                        obd->obd_name, &obd->obd_osfs,
1096                        obd->obd_osfs.os_bavail, obd->obd_osfs.os_blocks,
1097                        obd->obd_osfs.os_ffree, obd->obd_osfs.os_files);
1098                 spin_lock(&obd->obd_osfs_lock);
1099                 memcpy(osfs, &obd->obd_osfs, sizeof(*osfs));
1100                 spin_unlock(&obd->obd_osfs_lock);
1101         }
1102         RETURN(rc);
1103 }
1104
1105 static inline int obd_preprw(const struct lu_env *env, int cmd,
1106                              struct obd_export *exp, struct obdo *oa,
1107                              int objcount, struct obd_ioobj *obj,
1108                              struct niobuf_remote *remote, int *pages,
1109                              struct niobuf_local *local)
1110 {
1111         int rc;
1112
1113         ENTRY;
1114
1115         rc = exp_check_ops(exp);
1116         if (rc)
1117                 RETURN(rc);
1118
1119         if (!exp->exp_obd->obd_type->typ_dt_ops->o_preprw) {
1120                 CERROR("%s: no %s operation\n",
1121                        (exp)->exp_obd->obd_name, __func__);
1122                 RETURN(-ENOTSUPP);
1123         }
1124
1125         rc = OBP(exp->exp_obd, preprw)(env, cmd, exp, oa, objcount, obj, remote,
1126                                        pages, local);
1127
1128         RETURN(rc);
1129 }
1130
1131 static inline int obd_commitrw(const struct lu_env *env, int cmd,
1132                                struct obd_export *exp, struct obdo *oa,
1133                                int objcount, struct obd_ioobj *obj,
1134                                struct niobuf_remote *rnb, int pages,
1135                                struct niobuf_local *local, int rc)
1136 {
1137         ENTRY;
1138
1139         rc = exp_check_ops(exp);
1140         if (rc)
1141                 RETURN(rc);
1142
1143         if (!exp->exp_obd->obd_type->typ_dt_ops->o_commitrw) {
1144                 CERROR("%s: no %s operation\n",
1145                        (exp)->exp_obd->obd_name, __func__);
1146                 RETURN(-ENOTSUPP);
1147         }
1148
1149         rc = OBP(exp->exp_obd, commitrw)(env, cmd, exp, oa, objcount, obj,
1150                                          rnb, pages, local, rc);
1151
1152         RETURN(rc);
1153 }
1154
1155 static inline int obd_iocontrol(unsigned int cmd, struct obd_export *exp,
1156                                 int len, void *karg, void __user *uarg)
1157 {
1158         int rc;
1159         ENTRY;
1160
1161         rc = exp_check_ops(exp);
1162         if (rc)
1163                 RETURN(rc);
1164
1165         if (!exp->exp_obd->obd_type->typ_dt_ops->o_iocontrol) {
1166                 CERROR("%s: no %s operation\n",
1167                        (exp)->exp_obd->obd_name, __func__);
1168                 RETURN(-ENOTSUPP);
1169         }
1170
1171         rc = OBP(exp->exp_obd, iocontrol)(cmd, exp, len, karg, uarg);
1172         RETURN(rc);
1173 }
1174
1175 static inline void obd_import_event(struct obd_device *obd,
1176                                     struct obd_import *imp,
1177                                     enum obd_import_event event)
1178 {
1179         ENTRY;
1180         if (!obd) {
1181                 CERROR("NULL device\n");
1182                 EXIT;
1183                 return;
1184         }
1185
1186         if (obd->obd_set_up && OBP(obd, import_event))
1187                 OBP(obd, import_event)(obd, imp, event);
1188
1189         EXIT;
1190 }
1191
1192 static inline int obd_notify(struct obd_device *obd,
1193                              struct obd_device *watched,
1194                              enum obd_notify_event ev)
1195 {
1196         int rc;
1197         ENTRY;
1198         OBD_CHECK_DEV(obd);
1199
1200         if (!obd->obd_set_up) {
1201                 CDEBUG(D_HA, "obd %s not set up\n", obd->obd_name);
1202                 RETURN(-EINVAL);
1203         }
1204
1205         if (!OBP(obd, notify)) {
1206                 CDEBUG(D_HA, "obd %s has no notify handler\n", obd->obd_name);
1207                 RETURN(-ENOSYS);
1208         }
1209
1210         rc = OBP(obd, notify)(obd, watched, ev);
1211
1212         RETURN(rc);
1213 }
1214
1215 static inline int obd_notify_observer(struct obd_device *observer,
1216                                       struct obd_device *observed,
1217                                       enum obd_notify_event ev)
1218 {
1219         int rc = 0;
1220         int rc2 = 0;
1221         struct obd_notify_upcall *onu;
1222
1223         if (observer->obd_observer)
1224                 rc = obd_notify(observer->obd_observer, observed, ev);
1225
1226         /*
1227          * Also, call non-obd listener, if any
1228          */
1229         onu = &observer->obd_upcall;
1230         if (onu->onu_upcall != NULL)
1231                 rc2 = onu->onu_upcall(observer, observed, ev, onu->onu_owner);
1232
1233         return rc ? rc : rc2;
1234 }
1235
1236 static inline int obd_quotactl(struct obd_export *exp,
1237                                struct obd_quotactl *oqctl)
1238 {
1239         int rc;
1240         ENTRY;
1241
1242         rc = exp_check_ops(exp);
1243         if (rc)
1244                 RETURN(rc);
1245
1246         if (!exp->exp_obd->obd_type->typ_dt_ops->o_quotactl) {
1247                 CERROR("%s: no %s operation\n",
1248                        (exp)->exp_obd->obd_name, __func__);
1249                 RETURN(-ENOTSUPP);
1250         }
1251
1252         rc = OBP(exp->exp_obd, quotactl)(exp->exp_obd, exp, oqctl);
1253         RETURN(rc);
1254 }
1255
1256 static inline int obd_health_check(const struct lu_env *env,
1257                                    struct obd_device *obd)
1258 {
1259         /* returns: 0 on healthy
1260          *         >0 on unhealthy + reason code/flag
1261          *            however the only suppored reason == 1 right now
1262          *            We'll need to define some better reasons
1263          *            or flags in the future.
1264          *         <0 on error
1265          */
1266         int rc;
1267
1268         ENTRY;
1269
1270         /* NULL method is normal here */
1271         if (obd == NULL || !obd->obd_type) {
1272                 CERROR("cleaned up obd\n");
1273                 RETURN(-EOPNOTSUPP);
1274         }
1275         if (!obd->obd_set_up || obd->obd_stopping)
1276                 RETURN(0);
1277         if (!OBP(obd, health_check))
1278                 RETURN(0);
1279
1280         rc = OBP(obd, health_check)(env, obd);
1281         RETURN(rc);
1282 }
1283
1284 static inline int obd_register_observer(struct obd_device *obd,
1285                                         struct obd_device *observer)
1286 {
1287         ENTRY;
1288         OBD_CHECK_DEV(obd);
1289         down_write(&obd->obd_observer_link_sem);
1290         if (obd->obd_observer && observer) {
1291                 up_write(&obd->obd_observer_link_sem);
1292                 RETURN(-EALREADY);
1293         }
1294         obd->obd_observer = observer;
1295         up_write(&obd->obd_observer_link_sem);
1296         RETURN(0);
1297 }
1298
1299 /* metadata helpers */
1300 static inline int md_get_root(struct obd_export *exp, const char *fileset,
1301                               struct lu_fid *fid)
1302 {
1303         int rc;
1304
1305         ENTRY;
1306         rc = exp_check_ops(exp);
1307         if (rc)
1308                 RETURN(rc);
1309         EXP_MD_COUNTER_INCREMENT(exp, get_root);
1310         rc = MDP(exp->exp_obd, get_root)(exp, fileset, fid);
1311
1312         RETURN(rc);
1313 }
1314
1315 static inline int md_getattr(struct obd_export *exp,
1316                              struct md_op_data *op_data,
1317                              struct ptlrpc_request **request)
1318 {
1319         int rc;
1320
1321         ENTRY;
1322         rc = exp_check_ops(exp);
1323         if (rc)
1324                 RETURN(rc);
1325         EXP_MD_COUNTER_INCREMENT(exp, getattr);
1326         rc = MDP(exp->exp_obd, getattr)(exp, op_data, request);
1327         RETURN(rc);
1328 }
1329
1330 static inline int md_null_inode(struct obd_export *exp,
1331                                    const struct lu_fid *fid)
1332 {
1333         int rc;
1334         ENTRY;
1335         rc = exp_check_ops(exp);
1336         if (rc)
1337                 RETURN(rc);
1338         EXP_MD_COUNTER_INCREMENT(exp, null_inode);
1339         rc = MDP(exp->exp_obd, null_inode)(exp, fid);
1340         RETURN(rc);
1341 }
1342
1343 static inline int md_close(struct obd_export *exp, struct md_op_data *op_data,
1344                            struct md_open_data *mod,
1345                            struct ptlrpc_request **request)
1346 {
1347         int rc;
1348         ENTRY;
1349         rc = exp_check_ops(exp);
1350         if (rc)
1351                 RETURN(rc);
1352         EXP_MD_COUNTER_INCREMENT(exp, close);
1353         rc = MDP(exp->exp_obd, close)(exp, op_data, mod, request);
1354         RETURN(rc);
1355 }
1356
1357 static inline int md_create(struct obd_export *exp, struct md_op_data *op_data,
1358                             const void *data, size_t datalen, umode_t mode,
1359                             uid_t uid, gid_t gid, cfs_cap_t cap_effective,
1360                             __u64 rdev, struct ptlrpc_request **request)
1361 {
1362         int rc;
1363         ENTRY;
1364         rc = exp_check_ops(exp);
1365         if (rc)
1366                 RETURN(rc);
1367         EXP_MD_COUNTER_INCREMENT(exp, create);
1368         rc = MDP(exp->exp_obd, create)(exp, op_data, data, datalen, mode,
1369                                        uid, gid, cap_effective, rdev, request);
1370         RETURN(rc);
1371 }
1372
1373 static inline int md_enqueue(struct obd_export *exp,
1374                              struct ldlm_enqueue_info *einfo,
1375                              const union ldlm_policy_data *policy,
1376                              struct md_op_data *op_data,
1377                              struct lustre_handle *lockh,
1378                              __u64 extra_lock_flags)
1379 {
1380         int rc;
1381         ENTRY;
1382         rc = exp_check_ops(exp);
1383         if (rc)
1384                 RETURN(rc);
1385         EXP_MD_COUNTER_INCREMENT(exp, enqueue);
1386         rc = MDP(exp->exp_obd, enqueue)(exp, einfo, policy, op_data, lockh,
1387                                         extra_lock_flags);
1388         RETURN(rc);
1389 }
1390
1391 static inline int md_getattr_name(struct obd_export *exp,
1392                                   struct md_op_data *op_data,
1393                                   struct ptlrpc_request **request)
1394 {
1395         int rc;
1396         ENTRY;
1397         rc = exp_check_ops(exp);
1398         if (rc)
1399                 RETURN(rc);
1400         EXP_MD_COUNTER_INCREMENT(exp, getattr_name);
1401         rc = MDP(exp->exp_obd, getattr_name)(exp, op_data, request);
1402         RETURN(rc);
1403 }
1404
1405 static inline int md_intent_lock(struct obd_export *exp,
1406                                  struct md_op_data *op_data,
1407                                  struct lookup_intent *it,
1408                                  struct ptlrpc_request **reqp,
1409                                  ldlm_blocking_callback cb_blocking,
1410                                  __u64 extra_lock_flags)
1411 {
1412         int rc;
1413         ENTRY;
1414         rc = exp_check_ops(exp);
1415         if (rc)
1416                 RETURN(rc);
1417         EXP_MD_COUNTER_INCREMENT(exp, intent_lock);
1418         rc = MDP(exp->exp_obd, intent_lock)(exp, op_data, it, reqp, cb_blocking,
1419                                             extra_lock_flags);
1420         RETURN(rc);
1421 }
1422
1423 static inline int md_link(struct obd_export *exp, struct md_op_data *op_data,
1424                           struct ptlrpc_request **request)
1425 {
1426         int rc;
1427         ENTRY;
1428         rc = exp_check_ops(exp);
1429         if (rc)
1430                 RETURN(rc);
1431         EXP_MD_COUNTER_INCREMENT(exp, link);
1432         rc = MDP(exp->exp_obd, link)(exp, op_data, request);
1433         RETURN(rc);
1434 }
1435
1436 static inline int md_rename(struct obd_export *exp, struct md_op_data *op_data,
1437                             const char *old, size_t oldlen, const char *new,
1438                             size_t newlen, struct ptlrpc_request **request)
1439 {
1440         int rc;
1441         ENTRY;
1442         rc = exp_check_ops(exp);
1443         if (rc)
1444                 RETURN(rc);
1445
1446         EXP_MD_COUNTER_INCREMENT(exp, rename);
1447         rc = MDP(exp->exp_obd, rename)(exp, op_data, old, oldlen, new,
1448                                        newlen, request);
1449         RETURN(rc);
1450 }
1451
1452 static inline int md_setattr(struct obd_export *exp, struct md_op_data *op_data,
1453                              void *ea, size_t ealen,
1454                              struct ptlrpc_request **request)
1455 {
1456         int rc;
1457         ENTRY;
1458         rc = exp_check_ops(exp);
1459         if (rc)
1460                 RETURN(rc);
1461
1462         EXP_MD_COUNTER_INCREMENT(exp, setattr);
1463         rc = MDP(exp->exp_obd, setattr)(exp, op_data, ea, ealen, request);
1464         RETURN(rc);
1465 }
1466
1467 static inline int md_fsync(struct obd_export *exp, const struct lu_fid *fid,
1468                            struct ptlrpc_request **request)
1469 {
1470         int rc;
1471
1472         ENTRY;
1473         rc = exp_check_ops(exp);
1474         if (rc)
1475                 RETURN(rc);
1476
1477         EXP_MD_COUNTER_INCREMENT(exp, fsync);
1478         rc = MDP(exp->exp_obd, fsync)(exp, fid, request);
1479
1480         RETURN(rc);
1481 }
1482
1483 /* FLR: resync mirrored files. */
1484 static inline int md_file_resync(struct obd_export *exp,
1485                                  struct md_op_data *data)
1486 {
1487         int rc;
1488
1489         ENTRY;
1490         rc = exp_check_ops(exp);
1491         if (rc)
1492                 RETURN(rc);
1493
1494         EXP_MD_COUNTER_INCREMENT(exp, file_resync);
1495         rc = MDP(exp->exp_obd, file_resync)(exp, data);
1496
1497         RETURN(rc);
1498 }
1499
1500 static inline int md_read_page(struct obd_export *exp,
1501                                struct md_op_data *op_data,
1502                                struct md_callback *cb_op,
1503                                __u64  hash_offset,
1504                                struct page **ppage)
1505 {
1506         int rc;
1507         ENTRY;
1508         rc = exp_check_ops(exp);
1509         if (rc)
1510                 RETURN(rc);
1511
1512         EXP_MD_COUNTER_INCREMENT(exp, read_page);
1513         rc = MDP(exp->exp_obd, read_page)(exp, op_data, cb_op, hash_offset,
1514                                           ppage);
1515         RETURN(rc);
1516 }
1517
1518 static inline int md_unlink(struct obd_export *exp, struct md_op_data *op_data,
1519                             struct ptlrpc_request **request)
1520 {
1521         int rc;
1522         ENTRY;
1523         rc = exp_check_ops(exp);
1524         if (rc)
1525                 RETURN(rc);
1526
1527         EXP_MD_COUNTER_INCREMENT(exp, unlink);
1528         rc = MDP(exp->exp_obd, unlink)(exp, op_data, request);
1529         RETURN(rc);
1530 }
1531
1532 static inline int md_get_lustre_md(struct obd_export *exp,
1533                                    struct ptlrpc_request *req,
1534                                    struct obd_export *dt_exp,
1535                                    struct obd_export *md_exp,
1536                                    struct lustre_md *md)
1537 {
1538         int rc;
1539         ENTRY;
1540         rc = exp_check_ops(exp);
1541         if (rc)
1542                 RETURN(rc);
1543
1544         EXP_MD_COUNTER_INCREMENT(exp, get_lustre_md);
1545         RETURN(MDP(exp->exp_obd, get_lustre_md)(exp, req, dt_exp, md_exp, md));
1546 }
1547
1548 static inline int md_free_lustre_md(struct obd_export *exp,
1549                                     struct lustre_md *md)
1550 {
1551         int rc;
1552         ENTRY;
1553         rc = exp_check_ops(exp);
1554         if (rc)
1555                 RETURN(rc);
1556         EXP_MD_COUNTER_INCREMENT(exp, free_lustre_md);
1557         RETURN(MDP(exp->exp_obd, free_lustre_md)(exp, md));
1558 }
1559
1560 static inline int md_merge_attr(struct obd_export *exp,
1561                                 const struct lmv_stripe_md *lsm,
1562                                 struct cl_attr *attr,
1563                                 ldlm_blocking_callback cb)
1564 {
1565         int rc;
1566         ENTRY;
1567         rc = exp_check_ops(exp);
1568         if (rc)
1569                 RETURN(rc);
1570         EXP_MD_COUNTER_INCREMENT(exp, merge_attr);
1571         RETURN(MDP(exp->exp_obd, merge_attr)(exp, lsm, attr, cb));
1572 }
1573
1574 static inline int md_setxattr(struct obd_export *exp, const struct lu_fid *fid,
1575                               u64 obd_md_valid, const char *name,
1576                               const void *value, size_t value_size,
1577                               unsigned int xattr_flags, u32 suppgid,
1578                               struct ptlrpc_request **req)
1579 {
1580         int rc;
1581         ENTRY;
1582
1583         rc = exp_check_ops(exp);
1584         if (rc)
1585                 RETURN(rc);
1586         EXP_MD_COUNTER_INCREMENT(exp, setxattr);
1587         RETURN(MDP(exp->exp_obd, setxattr)(exp, fid, obd_md_valid, name,
1588                                            value, value_size, xattr_flags,
1589                                            suppgid, req));
1590 }
1591
1592 static inline int md_getxattr(struct obd_export *exp, const struct lu_fid *fid,
1593                               u64 obd_md_valid, const char *name,
1594                               size_t buf_size, struct ptlrpc_request **req)
1595 {
1596         int rc;
1597         ENTRY;
1598         rc = exp_check_ops(exp);
1599         if (rc)
1600                 RETURN(rc);
1601         EXP_MD_COUNTER_INCREMENT(exp, getxattr);
1602         RETURN(MDP(exp->exp_obd, getxattr)(exp, fid, obd_md_valid, name,
1603                                            buf_size, req));
1604 }
1605
1606 static inline int md_set_open_replay_data(struct obd_export *exp,
1607                                           struct obd_client_handle *och,
1608                                           struct lookup_intent *it)
1609 {
1610         int rc;
1611         ENTRY;
1612         rc = exp_check_ops(exp);
1613         if (rc)
1614                 RETURN(rc);
1615         EXP_MD_COUNTER_INCREMENT(exp, set_open_replay_data);
1616         RETURN(MDP(exp->exp_obd, set_open_replay_data)(exp, och, it));
1617 }
1618
1619 static inline int md_clear_open_replay_data(struct obd_export *exp,
1620                                             struct obd_client_handle *och)
1621 {
1622         int rc;
1623         ENTRY;
1624         rc = exp_check_ops(exp);
1625         if (rc)
1626                 RETURN(rc);
1627         EXP_MD_COUNTER_INCREMENT(exp, clear_open_replay_data);
1628         RETURN(MDP(exp->exp_obd, clear_open_replay_data)(exp, och));
1629 }
1630
1631 static inline int md_set_lock_data(struct obd_export *exp,
1632                                    const struct lustre_handle *lockh,
1633                                    void *data, __u64 *bits)
1634 {
1635         int rc;
1636         ENTRY;
1637         rc = exp_check_ops(exp);
1638         if (rc)
1639                 RETURN(rc);
1640         EXP_MD_COUNTER_INCREMENT(exp, set_lock_data);
1641         RETURN(MDP(exp->exp_obd, set_lock_data)(exp, lockh, data, bits));
1642 }
1643
1644 static inline
1645 int md_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
1646                      union ldlm_policy_data *policy, enum ldlm_mode mode,
1647                      enum ldlm_cancel_flags cancel_flags, void *opaque)
1648 {
1649         int rc;
1650         ENTRY;
1651
1652         rc = exp_check_ops(exp);
1653         if (rc)
1654                 RETURN(rc);
1655         EXP_MD_COUNTER_INCREMENT(exp, cancel_unused);
1656
1657         rc = MDP(exp->exp_obd, cancel_unused)(exp, fid, policy, mode,
1658                                               cancel_flags, opaque);
1659         RETURN(rc);
1660 }
1661
1662 static inline enum ldlm_mode md_lock_match(struct obd_export *exp, __u64 flags,
1663                                            const struct lu_fid *fid,
1664                                            enum ldlm_type type,
1665                                            union ldlm_policy_data *policy,
1666                                            enum ldlm_mode mode,
1667                                            struct lustre_handle *lockh)
1668 {
1669         int rc;
1670         ENTRY;
1671         rc = exp_check_ops(exp);
1672         if (rc)
1673                 RETURN(rc);
1674         EXP_MD_COUNTER_INCREMENT(exp, lock_match);
1675         RETURN(MDP(exp->exp_obd, lock_match)(exp, flags, fid, type,
1676                                              policy, mode, lockh));
1677 }
1678
1679 static inline int md_init_ea_size(struct obd_export *exp, __u32 ea_size,
1680                                   __u32 def_ea_size)
1681 {
1682         int rc;
1683         ENTRY;
1684         rc = exp_check_ops(exp);
1685         if (rc)
1686                 RETURN(rc);
1687         EXP_MD_COUNTER_INCREMENT(exp, init_ea_size);
1688         RETURN(MDP(exp->exp_obd, init_ea_size)(exp, ea_size, def_ea_size));
1689 }
1690
1691 static inline int md_intent_getattr_async(struct obd_export *exp,
1692                                           struct md_enqueue_info *minfo)
1693 {
1694         int rc;
1695         ENTRY;
1696         rc = exp_check_ops(exp);
1697         if (rc)
1698                 RETURN(rc);
1699         EXP_MD_COUNTER_INCREMENT(exp, intent_getattr_async);
1700         rc = MDP(exp->exp_obd, intent_getattr_async)(exp, minfo);
1701         RETURN(rc);
1702 }
1703
1704 static inline int md_revalidate_lock(struct obd_export *exp,
1705                                      struct lookup_intent *it,
1706                                      struct lu_fid *fid, __u64 *bits)
1707 {
1708         int rc;
1709         ENTRY;
1710         rc = exp_check_ops(exp);
1711         if (rc)
1712                 RETURN(rc);
1713         EXP_MD_COUNTER_INCREMENT(exp, revalidate_lock);
1714         rc = MDP(exp->exp_obd, revalidate_lock)(exp, it, fid, bits);
1715         RETURN(rc);
1716 }
1717
1718 static inline int md_get_fid_from_lsm(struct obd_export *exp,
1719                                       const struct lmv_stripe_md *lsm,
1720                                       const char *name, int namelen,
1721                                       struct lu_fid *fid)
1722 {
1723         int rc;
1724         ENTRY;
1725         rc = exp_check_ops(exp);
1726         if (rc)
1727                 RETURN(rc);
1728         EXP_MD_COUNTER_INCREMENT(exp, get_fid_from_lsm);
1729         rc = MDP(exp->exp_obd, get_fid_from_lsm)(exp, lsm, name, namelen, fid);
1730         RETURN(rc);
1731 }
1732
1733
1734 /* Unpack an MD struct from disk to in-memory format.
1735  * Returns +ve size of unpacked MD (0 for free), or -ve error.
1736  *
1737  * If *plsm != NULL and lmm == NULL then *lsm will be freed.
1738  * If *plsm == NULL then it will be allocated.
1739  */
1740 static inline int md_unpackmd(struct obd_export *exp,
1741                               struct lmv_stripe_md **plsm,
1742                               const union lmv_mds_md *lmm, size_t lmm_size)
1743 {
1744         int rc;
1745         ENTRY;
1746         rc = exp_check_ops(exp);
1747         if (rc)
1748                 RETURN(rc);
1749         EXP_MD_COUNTER_INCREMENT(exp, unpackmd);
1750         rc = MDP(exp->exp_obd, unpackmd)(exp, plsm, lmm, lmm_size);
1751         RETURN(rc);
1752 }
1753
1754 /* OBD Metadata Support */
1755
1756 extern int obd_init_caches(void);
1757 extern void obd_cleanup_caches(void);
1758
1759 /* support routines */
1760 extern struct kmem_cache *obdo_cachep;
1761
1762 #define OBDO_ALLOC(ptr)                                                       \
1763 do {                                                                          \
1764         OBD_SLAB_ALLOC_PTR_GFP((ptr), obdo_cachep, GFP_NOFS);             \
1765 } while(0)
1766
1767 #define OBDO_FREE(ptr)                                                        \
1768 do {                                                                          \
1769         OBD_SLAB_FREE_PTR((ptr), obdo_cachep);                                \
1770 } while(0)
1771
1772
1773 typedef int (*register_lwp_cb)(void *data);
1774
1775 struct lwp_register_item {
1776         struct obd_export **lri_exp;
1777         register_lwp_cb     lri_cb_func;
1778         void               *lri_cb_data;
1779         struct list_head    lri_list;
1780         atomic_t            lri_ref;
1781         char                lri_name[MTI_NAME_MAXLEN];
1782 };
1783
1784 /* I'm as embarrassed about this as you are.
1785  *
1786  * <shaver> // XXX do not look into _superhack with remaining eye
1787  * <shaver> // XXX if this were any uglier, I'd get my own show on MTV */
1788 extern int (*ptlrpc_put_connection_superhack)(struct ptlrpc_connection *c);
1789
1790 /* obd_mount.c */
1791 #ifdef HAVE_SERVER_SUPPORT
1792 int lustre_register_lwp_item(const char *lwpname, struct obd_export **exp,
1793                              register_lwp_cb cb_func, void *cb_data);
1794 void lustre_deregister_lwp_item(struct obd_export **exp);
1795 struct obd_export *lustre_find_lwp_by_index(const char *dev, __u32 idx);
1796 void lustre_notify_lwp_list(struct obd_export *exp);
1797 int tgt_name2lwp_name(const char *tgt_name, char *lwp_name, int len, __u32 idx);
1798 #endif /* HAVE_SERVER_SUPPORT */
1799 int lustre_register_fs(void);
1800 int lustre_unregister_fs(void);
1801 int lustre_check_exclusion(struct super_block *sb, char *svname);
1802
1803 typedef __u8 class_uuid_t[16];
1804 static inline void class_uuid_unparse(class_uuid_t uu, struct obd_uuid *out)
1805 {
1806         snprintf(out->uuid, sizeof(out->uuid), "%02x%02x%02x%02x-%02x%02x-"
1807                  "%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x",
1808                  uu[14], uu[15], uu[12], uu[13], uu[10], uu[11], uu[8], uu[9],
1809                  uu[6], uu[7], uu[4], uu[5], uu[2], uu[3], uu[0], uu[1]);
1810 }
1811
1812 /* lustre_peer.c    */
1813 int lustre_uuid_to_peer(const char *uuid, lnet_nid_t *peer_nid, int index);
1814 int class_add_uuid(const char *uuid, __u64 nid);
1815 int class_del_uuid (const char *uuid);
1816 int class_check_uuid(struct obd_uuid *uuid, __u64 nid);
1817
1818 /* class_obd.c */
1819 extern char obd_jobid_name[];
1820
1821 /* prng.c */
1822 #define ll_generate_random_uuid(uuid_out) cfs_get_random_bytes(uuid_out, sizeof(class_uuid_t))
1823
1824 /* statfs_pack.c */
1825 struct kstatfs;
1826 void statfs_pack(struct obd_statfs *osfs, struct kstatfs *sfs);
1827 void statfs_unpack(struct kstatfs *sfs, struct obd_statfs *osfs);
1828
1829 /* root squash info */
1830 struct rw_semaphore;
1831 struct root_squash_info {
1832         uid_t                   rsi_uid;
1833         gid_t                   rsi_gid;
1834         struct list_head        rsi_nosquash_nids;
1835         struct rw_semaphore     rsi_sem;
1836 };
1837
1838 int server_name2index(const char *svname, __u32 *idx, const char **endptr);
1839
1840 /* linux-module.c */
1841 extern struct miscdevice obd_psdev;
1842 int obd_ioctl_getdata(char **buf, int *len, void __user *arg);
1843 int class_procfs_init(void);
1844 int class_procfs_clean(void);
1845 #endif /* __LINUX_OBD_CLASS_H */