Whamcloud - gitweb
e154556e0f679de18863769f5ef0d340ee3b3cf0
[fs/lustre-release.git] / lustre / osp / osp_dev.c
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) 2012, 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  * lustre/osp/osp_dev.c
33  *
34  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
35  * Author: Mikhail Pershin <mike.pershin@intel.com>
36  * Author: Di Wang <di.wang@intel.com>
37  */
38 /*
39  * The Object Storage Proxy (OSP) module provides an implementation of
40  * the DT API for remote MDTs and OSTs. Every local OSP device (or
41  * object) is a proxy for a remote OSD device (or object). Thus OSP
42  * converts DT operations into RPCs, which are sent to the OUT service
43  * on a remote target, converted back to DT operations, and
44  * executed. Of course there are many ways in which this description
45  * is inaccurate but it's a good enough mental model. OSP is used by
46  * the MDT stack in several ways:
47  *
48  * - OSP devices allocate FIDs for the stripe sub-objects of a striped
49  *   file or directory.
50  *
51  * - OSP objects represent the remote MDT and OST objects that are
52  *   the stripes of a striped object.
53  *
54  * - OSP devices log, send, and track synchronous operations (setattr
55  *   and unlink) to remote targets.
56  *
57  * - OSP objects are the bottom slice of the compound LU object
58  *   representing a remote MDT object: MDT/MDD/LOD/OSP.
59  *
60  * - OSP objects are used by LFSCK to represent remote OST objects
61  *   during the verification of MDT-OST consistency.
62  *
63  * - OSP devices batch idempotent requests (declare_attr_get() and
64  *   declare_xattr_get()) to the remote target and cache their results.
65  *
66  * In addition the OSP layer implements a subset of the OBD device API
67  * to support being a client of a remote target, connecting to other
68  * layers, and FID allocation.
69  */
70
71 #define DEBUG_SUBSYSTEM S_MDS
72
73 #include <linux/kthread.h>
74
75 #include <uapi/linux/lustre/lustre_ioctl.h>
76 #include <lustre_log.h>
77 #include <lustre_obdo.h>
78 #include <uapi/linux/lustre/lustre_param.h>
79 #include <obd_class.h>
80
81 #include "osp_internal.h"
82
83 /* Slab for OSP object allocation */
84 struct kmem_cache *osp_object_kmem;
85
86 static struct lu_kmem_descr osp_caches[] = {
87         {
88                 .ckd_cache = &osp_object_kmem,
89                 .ckd_name  = "osp_obj",
90                 .ckd_size  = sizeof(struct osp_object)
91         },
92         {
93                 .ckd_cache = NULL
94         }
95 };
96
97 /**
98  * Implementation of lu_device_operations::ldo_object_alloc
99  *
100  * Allocates an OSP object in memory, whose FID is on the remote target.
101  *
102  * \param[in] env       execution environment
103  * \param[in] hdr       The header of the object stack. If it is NULL, it
104  *                      means the object is not built from top device, i.e.
105  *                      it is a sub-stripe object of striped directory or
106  *                      an OST object.
107  * \param[in] d         OSP device
108  *
109  * \retval object       object being created if the creation succeed.
110  * \retval NULL         NULL if the creation failed.
111  */
112 static struct lu_object *osp_object_alloc(const struct lu_env *env,
113                                           const struct lu_object_header *hdr,
114                                           struct lu_device *d)
115 {
116         struct lu_object_header *h = NULL;
117         struct osp_object       *o;
118         struct lu_object        *l;
119
120         OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, GFP_NOFS);
121         if (o != NULL) {
122                 l = &o->opo_obj.do_lu;
123
124                 /* If hdr is NULL, it means the object is not built
125                  * from the top dev(MDT/OST), usually it happens when
126                  * building striped object, like data object on MDT or
127                  * striped object for directory */
128                 if (hdr == NULL) {
129                         h = &o->opo_header;
130                         lu_object_header_init(h);
131                         dt_object_init(&o->opo_obj, h, d);
132                         lu_object_add_top(h, l);
133                 } else {
134                         dt_object_init(&o->opo_obj, h, d);
135                 }
136
137                 l->lo_ops = &osp_lu_obj_ops;
138
139                 return l;
140         } else {
141                 return NULL;
142         }
143 }
144
145 /**
146  * Find or create the local object
147  *
148  * Finds or creates the local file referenced by \a reg_id and return the
149  * attributes of the local file.
150  *
151  * \param[in] env       execution environment
152  * \param[in] osp       OSP device
153  * \param[out] attr     attributes of the object
154  * \param[in] reg_id    the local object ID of the file. It will be used
155  *                      to compose a local FID{FID_SEQ_LOCAL_FILE, reg_id, 0}
156  *                      to identify the object.
157  *
158  * \retval object               object(dt_object) found or created
159  * \retval ERR_PTR(errno)       ERR_PTR(errno) if not get the object.
160  */
161 static struct dt_object
162 *osp_find_or_create_local_file(const struct lu_env *env, struct osp_device *osp,
163                                struct lu_attr *attr, __u32 reg_id)
164 {
165         struct osp_thread_info *osi = osp_env_info(env);
166         struct dt_object_format dof = { 0 };
167         struct dt_object       *dto;
168         int                  rc;
169         ENTRY;
170
171         lu_local_obj_fid(&osi->osi_fid, reg_id);
172         attr->la_valid = LA_MODE;
173         attr->la_mode = S_IFREG | 0644;
174         dof.dof_type = DFT_REGULAR;
175         /* Find or create the local object by osi_fid. */
176         dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
177                                 &dof, attr);
178         if (IS_ERR(dto))
179                 RETURN(dto);
180
181         /* Get attributes of the local object. */
182         rc = dt_attr_get(env, dto, attr);
183         if (rc) {
184                 CERROR("%s: can't be initialized: rc = %d\n",
185                        osp->opd_obd->obd_name, rc);
186                 dt_object_put(env, dto);
187                 RETURN(ERR_PTR(rc));
188         }
189         RETURN(dto);
190 }
191
192 /**
193  * Write data buffer to a local file object.
194  *
195  * \param[in] env       execution environment
196  * \param[in] osp       OSP device
197  * \param[in] dt_obj    object written to
198  * \param[in] buf       buffer containing byte array and length
199  * \param[in] offset    write offset in the object in bytes
200  *
201  * \retval 0            0 if write succeed
202  * \retval -EFAULT      -EFAULT if only part of buffer is written.
203  * \retval negative             other negative errno if write failed.
204  */
205 static int osp_write_local_file(const struct lu_env *env,
206                                 struct osp_device *osp,
207                                 struct dt_object *dt_obj,
208                                 struct lu_buf *buf,
209                                 loff_t offset)
210 {
211         struct thandle *th;
212         int rc;
213
214         if (osp->opd_storage->dd_rdonly)
215                 RETURN(0);
216
217         th = dt_trans_create(env, osp->opd_storage);
218         if (IS_ERR(th))
219                 RETURN(PTR_ERR(th));
220
221         rc = dt_declare_record_write(env, dt_obj, buf, offset, th);
222         if (rc)
223                 GOTO(out, rc);
224         rc = dt_trans_start_local(env, osp->opd_storage, th);
225         if (rc)
226                 GOTO(out, rc);
227
228         rc = dt_record_write(env, dt_obj, buf, &offset, th);
229 out:
230         dt_trans_stop(env, osp->opd_storage, th);
231         RETURN(rc);
232 }
233
234 /**
235  * Initialize last ID object.
236  *
237  * This function initializes the LAST_ID file, which stores the current last
238  * used id of data objects. The MDT will use the last used id and the last_seq
239  * (\see osp_init_last_seq()) to synchronize the precreate object cache with
240  * OSTs.
241  *
242  * \param[in] env       execution environment
243  * \param[in] osp       OSP device
244  *
245  * \retval 0            0 if initialization succeed
246  * \retval negative     negative errno if initialization failed
247  */
248 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
249 {
250         struct osp_thread_info  *osi = osp_env_info(env);
251         struct lu_fid           *fid = &osp->opd_last_used_fid;
252         struct dt_object        *dto;
253         int                     rc = -EFAULT;
254         ENTRY;
255
256         dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
257                                             MDD_LOV_OBJ_OID);
258         if (IS_ERR(dto))
259                 RETURN(PTR_ERR(dto));
260
261         osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &osp->opd_last_id,
262                            osp->opd_index);
263
264         /* object will be released in device cleanup path */
265         if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
266                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
267                 if (rc != 0 && rc != -EFAULT)
268                         GOTO(out, rc);
269                 /* In case of idif bits 32-48 go to f_seq
270                  * (see osp_init_last_seq). So don't care
271                  * about u64->u32 convertion. */
272                 fid->f_oid = osp->opd_last_id;
273         }
274
275         if (rc == -EFAULT) { /* fresh LAST_ID */
276                 osp->opd_last_id = 0;
277                 fid->f_oid = 0;
278                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
279                                           osi->osi_off);
280                 if (rc != 0)
281                         GOTO(out, rc);
282         }
283         osp->opd_last_used_oid_file = dto;
284         RETURN(0);
285 out:
286         /* object will be released in device cleanup path */
287         CERROR("%s: can't initialize lov_objid: rc = %d\n",
288                osp->opd_obd->obd_name, rc);
289         dt_object_put(env, dto);
290         osp->opd_last_used_oid_file = NULL;
291         RETURN(rc);
292 }
293
294 /**
295  * Initialize last sequence object.
296  *
297  * This function initializes the LAST_SEQ file in the local OSD, which stores
298  * the current last used sequence of data objects. The MDT will use the last
299  * sequence and last id (\see osp_init_last_objid()) to synchronize the
300  * precreate object cache with OSTs.
301  *
302  * \param[in] env       execution environment
303  * \param[in] osp       OSP device
304  *
305  * \retval 0            0 if initialization succeed
306  * \retval negative     negative errno if initialization failed
307  */
308 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
309 {
310         struct osp_thread_info  *osi = osp_env_info(env);
311         struct lu_fid           *fid = &osp->opd_last_used_fid;
312         struct dt_object        *dto;
313         int                     rc = -EFAULT;
314         ENTRY;
315
316         dto = osp_find_or_create_local_file(env, osp, &osi->osi_attr,
317                                             MDD_LOV_OBJ_OSEQ);
318         if (IS_ERR(dto))
319                 RETURN(PTR_ERR(dto));
320
321         osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
322                            osp->opd_index);
323
324         /* object will be released in device cleanup path */
325         if (osi->osi_attr.la_size >= (osi->osi_off + osi->osi_lb.lb_len)) {
326                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
327                 if (rc != 0 && rc != -EFAULT)
328                         GOTO(out, rc);
329                 if (fid_is_idif(fid))
330                         fid->f_seq = fid_idif_seq(osp->opd_last_id,
331                                                   osp->opd_index);
332         }
333
334         if (rc == -EFAULT) { /* fresh OSP */
335                 fid->f_seq = 0;
336                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
337                                           osi->osi_off);
338                 if (rc != 0)
339                         GOTO(out, rc);
340         }
341         osp->opd_last_used_seq_file = dto;
342         RETURN(0);
343 out:
344         /* object will be released in device cleanup path */
345         CERROR("%s: can't initialize lov_seq: rc = %d\n",
346                osp->opd_obd->obd_name, rc);
347         dt_object_put(env, dto);
348         osp->opd_last_used_seq_file = NULL;
349         RETURN(rc);
350 }
351
352 /**
353  * Initialize last OID and sequence object.
354  *
355  * If the MDT is just upgraded to 2.4 from the lower version, where the
356  * LAST_SEQ file does not exist, the file will be created and IDIF sequence
357  * will be written into the file.
358  *
359  * \param[in] env       execution environment
360  * \param[in] osp       OSP device
361  *
362  * \retval 0            0 if initialization succeed
363  * \retval negative     negative error if initialization failed
364  */
365 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
366 {
367         struct osp_thread_info *osi = osp_env_info(env);
368         int                  rc;
369         ENTRY;
370
371         fid_zero(&osp->opd_last_used_fid);
372         rc = osp_init_last_objid(env, osp);
373         if (rc < 0) {
374                 CERROR("%s: Can not get ids %d from old objid!\n",
375                        osp->opd_obd->obd_name, rc);
376                 RETURN(rc);
377         }
378
379         rc = osp_init_last_seq(env, osp);
380         if (rc < 0) {
381                 CERROR("%s: Can not get sequence %d from old objseq!\n",
382                        osp->opd_obd->obd_name, rc);
383                 GOTO(out, rc);
384         }
385
386         if (fid_oid(&osp->opd_last_used_fid) != 0 &&
387             fid_seq(&osp->opd_last_used_fid) == 0) {
388                 /* Just upgrade from the old version,
389                  * set the seq to be IDIF */
390                 osp->opd_last_used_fid.f_seq =
391                    fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
392                                 osp->opd_index);
393                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
394                                     &osp->opd_last_used_fid.f_seq,
395                                     osp->opd_index);
396                 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
397                                           &osi->osi_lb, osi->osi_off);
398                 if (rc) {
399                         CERROR("%s : Can not write seq file: rc = %d\n",
400                                osp->opd_obd->obd_name, rc);
401                         GOTO(out, rc);
402                 }
403         }
404
405         if (!fid_is_zero(&osp->opd_last_used_fid) &&
406                  !fid_is_sane(&osp->opd_last_used_fid)) {
407                 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
408                         PFID(&osp->opd_last_used_fid));
409                 GOTO(out, rc = -EINVAL);
410         }
411
412         osp_fid_to_obdid(&osp->opd_last_used_fid, &osp->opd_last_id);
413         CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
414                osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
415 out:
416         if (rc != 0) {
417                 if (osp->opd_last_used_oid_file != NULL) {
418                         dt_object_put(env, osp->opd_last_used_oid_file);
419                         osp->opd_last_used_oid_file = NULL;
420                 }
421                 if (osp->opd_last_used_seq_file != NULL) {
422                         dt_object_put(env, osp->opd_last_used_seq_file);
423                         osp->opd_last_used_seq_file = NULL;
424                 }
425         }
426
427         RETURN(rc);
428 }
429
430 /**
431  * Release the last sequence and OID file objects in OSP device.
432  *
433  * \param[in] env       execution environment
434  * \param[in] osp       OSP device
435  */
436 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *osp)
437 {
438         /* release last_used file */
439         if (osp->opd_last_used_oid_file != NULL) {
440                 dt_object_put(env, osp->opd_last_used_oid_file);
441                 osp->opd_last_used_oid_file = NULL;
442         }
443
444         if (osp->opd_last_used_seq_file != NULL) {
445                 dt_object_put(env, osp->opd_last_used_seq_file);
446                 osp->opd_last_used_seq_file = NULL;
447         }
448 }
449
450 /**
451  * Disconnects the connection between OSP and its correspondent MDT or OST, and
452  * the import will be marked as inactive. It will only be called during OSP
453  * cleanup process.
454  *
455  * \param[in] d         OSP device being disconnected
456  *
457  * \retval 0            0 if disconnection succeed
458  * \retval negative     negative errno if disconnection failed
459  */
460 static int osp_disconnect(struct osp_device *d)
461 {
462         struct obd_device *obd = d->opd_obd;
463         struct obd_import *imp;
464         int rc = 0;
465
466         imp = obd->u.cli.cl_import;
467
468         /* Mark import deactivated now, so we don't try to reconnect if any
469          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
470          * fully deactivate the import, or that would drop all requests. */
471         LASSERT(imp != NULL);
472         spin_lock(&imp->imp_lock);
473         imp->imp_deactive = 1;
474         spin_unlock(&imp->imp_lock);
475
476         ptlrpc_deactivate_import(imp);
477
478         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
479          * delete it regardless.  (It's safe to delete an import that was
480          * never added.) */
481         (void)ptlrpc_pinger_del_import(imp);
482
483         rc = ptlrpc_disconnect_import(imp, 0);
484         if (rc != 0)
485                 CERROR("%s: can't disconnect: rc = %d\n", obd->obd_name, rc);
486
487         ptlrpc_invalidate_import(imp);
488
489         RETURN(rc);
490 }
491
492 /**
493  * Initialize the osp_update structure in OSP device
494  *
495  * Allocate osp update structure and start update thread.
496  *
497  * \param[in] osp       OSP device
498  *
499  * \retval              0 if initialization succeeds.
500  * \retval              negative errno if initialization fails.
501  */
502 static int osp_update_init(struct osp_device *osp)
503 {
504         struct task_struct *task;
505         int rc;
506
507         ENTRY;
508
509         LASSERT(osp->opd_connect_mdt);
510
511         if (osp->opd_storage->dd_rdonly)
512                 RETURN(0);
513
514         OBD_ALLOC_PTR(osp->opd_update);
515         if (osp->opd_update == NULL)
516                 RETURN(-ENOMEM);
517
518         init_waitqueue_head(&osp->opd_update->ou_waitq);
519         spin_lock_init(&osp->opd_update->ou_lock);
520         INIT_LIST_HEAD(&osp->opd_update->ou_list);
521         osp->opd_update->ou_rpc_version = 1;
522         osp->opd_update->ou_version = 1;
523         osp->opd_update->ou_generation = 0;
524
525         rc = lu_env_init(&osp->opd_update->ou_env,
526                          osp->opd_dt_dev.dd_lu_dev.ld_type->ldt_ctx_tags);
527         if (rc < 0) {
528                 CERROR("%s: init env error: rc = %d\n", osp->opd_obd->obd_name,
529                        rc);
530                 OBD_FREE_PTR(osp->opd_update);
531                 osp->opd_update = NULL;
532                 RETURN(rc);
533         }
534         /* start thread handling sending updates to the remote MDT */
535         task = kthread_create(osp_send_update_thread, osp,
536                               "osp_up%u-%u", osp->opd_index, osp->opd_group);
537         if (IS_ERR(task)) {
538                 int rc = PTR_ERR(task);
539
540                 lu_env_fini(&osp->opd_update->ou_env);
541                 OBD_FREE_PTR(osp->opd_update);
542                 osp->opd_update = NULL;
543                 CERROR("%s: can't start precreate thread: rc = %d\n",
544                        osp->opd_obd->obd_name, rc);
545                 RETURN(rc);
546         }
547
548         osp->opd_update->ou_update_task = task;
549         wake_up_process(task);
550
551         RETURN(0);
552 }
553
554 /**
555  * Finialize osp_update structure in OSP device
556  *
557  * Stop the OSP update sending thread, then delete the left
558  * osp thandle in the sending list.
559  *
560  * \param [in] osp      OSP device.
561  */
562 static void osp_update_fini(const struct lu_env *env, struct osp_device *osp)
563 {
564         struct osp_update_request *our;
565         struct osp_update_request *tmp;
566         struct osp_updates *ou = osp->opd_update;
567
568         if (ou == NULL)
569                 return;
570
571         kthread_stop(ou->ou_update_task);
572         lu_env_fini(&ou->ou_env);
573
574         /* Remove the left osp thandle from the list */
575         spin_lock(&ou->ou_lock);
576         list_for_each_entry_safe(our, tmp, &ou->ou_list,
577                                  our_list) {
578                 list_del_init(&our->our_list);
579                 LASSERT(our->our_th != NULL);
580                 osp_trans_callback(env, our->our_th, -EIO);
581                 /* our will be destroyed in osp_thandle_put() */
582                 osp_thandle_put(env, our->our_th);
583         }
584         spin_unlock(&ou->ou_lock);
585
586         OBD_FREE_PTR(ou);
587         osp->opd_update = NULL;
588 }
589
590 /**
591  * Cleanup OSP, which includes disconnect import, cleanup unlink log, stop
592  * precreate threads etc.
593  *
594  * \param[in] env       execution environment.
595  * \param[in] d         OSP device being disconnected.
596  *
597  * \retval 0            0 if cleanup succeed
598  * \retval negative     negative errno if cleanup failed
599  */
600 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
601 {
602         int                      rc = 0;
603         ENTRY;
604
605         LASSERT(env);
606
607         rc = osp_disconnect(d);
608
609         osp_statfs_fini(d);
610
611         if (!d->opd_connect_mdt) {
612                 /* stop sync thread */
613                 osp_sync_fini(d);
614
615                 /* stop precreate thread */
616                 osp_precreate_fini(d);
617
618                 /* release last_used file */
619                 osp_last_used_fini(env, d);
620         }
621
622         obd_fid_fini(d->opd_obd);
623
624         RETURN(rc);
625 }
626
627 /**
628  * Implementation of osp_lu_ops::ldo_process_config
629  *
630  * This function processes config log records in OSP layer. It is usually
631  * called from the top layer of MDT stack, and goes through the stack by calling
632  * ldo_process_config of next layer.
633  *
634  * \param[in] env       execution environment
635  * \param[in] dev       lu_device of OSP
636  * \param[in] lcfg      config log
637  *
638  * \retval 0            0 if the config log record is executed correctly.
639  * \retval negative     negative errno if the record execution fails.
640  */
641 static int osp_process_config(const struct lu_env *env,
642                               struct lu_device *dev, struct lustre_cfg *lcfg)
643 {
644         struct osp_device *d = lu2osp_dev(dev);
645         struct dt_device *dt = lu2dt_dev(dev);
646         struct obd_device *obd = d->opd_obd;
647         ssize_t count;
648         int rc;
649
650         ENTRY;
651
652         switch (lcfg->lcfg_command) {
653         case LCFG_PRE_CLEANUP:
654                 rc = osp_disconnect(d);
655                 osp_update_fini(env, d);
656                 if (obd->obd_namespace != NULL)
657                         ldlm_namespace_free_prior(obd->obd_namespace, NULL, 1);
658                 break;
659         case LCFG_CLEANUP:
660                 lu_dev_del_linkage(dev->ld_site, dev);
661                 rc = osp_shutdown(env, d);
662                 break;
663         case LCFG_PARAM:
664                 count = class_modify_config(lcfg, d->opd_connect_mdt ?
665                                                   PARAM_OSP : PARAM_OSC,
666                                             &dt->dd_kobj);
667                 if (count < 0) {
668                         /* class_modify_config() haven't found matching
669                          * parameter and returned an error so that layer(s)
670                          * below could use that. But OSP is the bottom, so
671                          * just ignore it
672                          */
673                         CERROR("%s: unknown param %s\n",
674                                (char *)lustre_cfg_string(lcfg, 0),
675                                (char *)lustre_cfg_string(lcfg, 1));
676                 }
677                 rc = 0;
678                 break;
679         default:
680                 CERROR("%s: unknown command %u\n",
681                        (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
682                 rc = 0;
683                 break;
684         }
685
686         RETURN(rc);
687 }
688
689 /**
690  * Implementation of osp_lu_ops::ldo_recovery_complete
691  *
692  * This function is called after recovery is finished, and OSP layer
693  * will wake up precreate thread here.
694  *
695  * \param[in] env       execution environment
696  * \param[in] dev       lu_device of OSP
697  *
698  * \retval 0            0 unconditionally
699  */
700 static int osp_recovery_complete(const struct lu_env *env,
701                                  struct lu_device *dev)
702 {
703         struct osp_device       *osp = lu2osp_dev(dev);
704
705         ENTRY;
706         osp->opd_recovery_completed = 1;
707
708         if (!osp->opd_connect_mdt && osp->opd_pre != NULL)
709                 wake_up(&osp->opd_pre_waitq);
710
711         RETURN(0);
712 }
713
714 /**
715  * Implementation of lu_device_operations::ldo_fid_alloc() for OSP
716  *
717  * Allocate FID from remote MDT.
718  *
719  * see include/lu_object.h for the details.
720  */
721 static int osp_fid_alloc(const struct lu_env *env, struct lu_device *d,
722                          struct lu_fid *fid, struct lu_object *parent,
723                          const struct lu_name *name)
724 {
725         struct osp_device *osp = lu2osp_dev(d);
726         struct client_obd *cli = &osp->opd_obd->u.cli;
727         struct lu_client_seq *seq = cli->cl_seq;
728         int rc;
729
730         ENTRY;
731
732         /* Sigh, fid client is not ready yet */
733         if (!osp->opd_obd->u.cli.cl_seq)
734                 RETURN(-ENOTCONN);
735
736         if (!osp->opd_obd->u.cli.cl_seq->lcs_exp)
737                 RETURN(-ENOTCONN);
738
739         rc = seq_client_alloc_fid(env, seq, fid);
740
741         RETURN(rc);
742 }
743
744 const struct lu_device_operations osp_lu_ops = {
745         .ldo_object_alloc       = osp_object_alloc,
746         .ldo_process_config     = osp_process_config,
747         .ldo_recovery_complete  = osp_recovery_complete,
748         .ldo_fid_alloc          = osp_fid_alloc,
749 };
750
751 /**
752  * Implementation of dt_device_operations::dt_statfs
753  *
754  * This function provides statfs status (for precreation) from
755  * corresponding OST. Note: this function only retrieves the status
756  * from the OSP device, and the real statfs RPC happens inside
757  * precreate thread (\see osp_statfs_update). Note: OSP for MDT does
758  * not need to retrieve statfs data for now.
759  *
760  * \param[in] env       execution environment.
761  * \param[in] dev       dt_device of OSP.
762  * \param[out] sfs      holds the retrieved statfs data.
763  *
764  * \retval 0            0 statfs data was retrieved successfully or
765  *                      retrieval was not needed
766  * \retval negative     negative errno if get statfs failed.
767  */
768 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
769                       struct obd_statfs *sfs, struct obd_statfs_info *info)
770 {
771         struct osp_device *d = dt2osp_dev(dev);
772         struct obd_import *imp = d->opd_obd->u.cli.cl_import;
773
774         ENTRY;
775
776         if (imp->imp_state == LUSTRE_IMP_CLOSED)
777                 RETURN(-ESHUTDOWN);
778
779         if (unlikely(d->opd_imp_active == 0))
780                 RETURN(-ENOTCONN);
781
782         /* return recently updated data */
783         *sfs = d->opd_statfs;
784         if (info) {
785                 info->os_reserved_mb_low = d->opd_reserved_mb_low;
786                 info->os_reserved_mb_high = d->opd_reserved_mb_high;
787         }
788
789         if (d->opd_pre == NULL)
790                 RETURN(0);
791
792         CDEBUG(D_OTHER,
793                "%s: %llu blocks, %llu free, %llu avail, %u reserved mb low, %u reserved mb high, %llu files, %llu free files\n",
794                d->opd_obd->obd_name,
795                sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
796                d->opd_reserved_mb_low, d->opd_reserved_mb_high,
797                sfs->os_files, sfs->os_ffree);
798
799
800         if (info && !info->os_enable_pre)
801                 RETURN(0);
802
803         /*
804          * The layer above osp (usually lod) can use f_precreated to
805          * estimate how many objects are available for immediate usage.
806          */
807         spin_lock(&d->opd_pre_lock);
808         sfs->os_fprecreated = osp_fid_diff(&d->opd_pre_last_created_fid,
809                                            &d->opd_pre_used_fid);
810         sfs->os_fprecreated -= d->opd_pre_reserved;
811         LASSERTF(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2,
812                  "last_created "DFID", next_fid "DFID", reserved %llu\n",
813                  PFID(&d->opd_pre_last_created_fid), PFID(&d->opd_pre_used_fid),
814                  d->opd_pre_reserved);
815         spin_unlock(&d->opd_pre_lock);
816         RETURN(0);
817 }
818
819 /**
820  * Implementation of dt_device_operations::dt_sync
821  *
822  * This function synchronizes the OSP cache to the remote target. It wakes
823  * up unlink log threads and sends out unlink records to the remote OST.
824  *
825  * \param[in] env       execution environment
826  * \param[in] dev       dt_device of OSP
827  *
828  * \retval 0            0 if synchronization succeeds
829  * \retval negative     negative errno if synchronization fails
830  */
831 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
832 {
833         struct osp_device *d = dt2osp_dev(dev);
834         time64_t start = ktime_get_seconds();
835         int recs, rc = 0;
836         u64 old;
837
838         ENTRY;
839
840         /* No Sync between MDTs yet. */
841         if (d->opd_connect_mdt)
842                 RETURN(0);
843
844         recs = atomic_read(&d->opd_sync_changes);
845         old = atomic64_read(&d->opd_sync_processed_recs);
846
847         osp_sync_force(env, dt2osp_dev(dev));
848
849         if (unlikely(d->opd_imp_active == 0))
850                 RETURN(-ENOTCONN);
851
852         down_write(&d->opd_async_updates_rwsem);
853
854         CDEBUG(D_OTHER, "%s: async updates %d\n", d->opd_obd->obd_name,
855                atomic_read(&d->opd_async_updates_count));
856
857         /* make sure the connection is fine */
858         rc = wait_event_idle_timeout(
859                 d->opd_sync_barrier_waitq,
860                 atomic_read(&d->opd_async_updates_count) == 0,
861                 cfs_time_seconds(obd_timeout));
862         if (rc > 0)
863                 rc = 0;
864         else if (rc == 0)
865                 rc = -ETIMEDOUT;
866
867         up_write(&d->opd_async_updates_rwsem);
868         if (rc != 0)
869                 GOTO(out, rc);
870
871         CDEBUG(D_CACHE, "%s: processed %llu\n", d->opd_obd->obd_name,
872                (unsigned long long)atomic64_read(&d->opd_sync_processed_recs));
873
874         while (atomic64_read(&d->opd_sync_processed_recs) < old + recs) {
875                 __u64 last = atomic64_read(&d->opd_sync_processed_recs);
876                 /* make sure the connection is fine */
877                 wait_event_idle_timeout(
878                         d->opd_sync_barrier_waitq,
879                         atomic64_read(&d->opd_sync_processed_recs)
880                              >= old + recs,
881                         cfs_time_seconds(obd_timeout));
882
883                 if (atomic64_read(&d->opd_sync_processed_recs) >= old + recs)
884                         break;
885
886                 if (atomic64_read(&d->opd_sync_processed_recs) != last) {
887                         /* some progress have been made,
888                          * keep trying... */
889                         continue;
890                 }
891
892                 /* no changes and expired, something is wrong */
893                 GOTO(out, rc = -ETIMEDOUT);
894         }
895
896         /* block new processing (barrier>0 - few callers are possible */
897         atomic_inc(&d->opd_sync_barrier);
898
899         CDEBUG(D_CACHE, "%s: %u in flight\n", d->opd_obd->obd_name,
900                atomic_read(&d->opd_sync_rpcs_in_flight));
901
902         /* wait till all-in-flight are replied, so executed by the target */
903         /* XXX: this is used by LFSCK at the moment, which doesn't require
904          *      all the changes to be committed, but in general it'd be
905          *      better to wait till commit */
906         while (atomic_read(&d->opd_sync_rpcs_in_flight) > 0) {
907                 old = atomic_read(&d->opd_sync_rpcs_in_flight);
908
909                 wait_event_idle_timeout(
910                         d->opd_sync_barrier_waitq,
911                         atomic_read(&d->opd_sync_rpcs_in_flight) == 0,
912                         cfs_time_seconds(obd_timeout));
913
914                 if (atomic_read(&d->opd_sync_rpcs_in_flight) == 0)
915                         break;
916
917                 if (atomic_read(&d->opd_sync_rpcs_in_flight) != old) {
918                         /* some progress have been made */
919                         continue;
920                 }
921
922                 /* no changes and expired, something is wrong */
923                 GOTO(out, rc = -ETIMEDOUT);
924         }
925
926 out:
927         /* resume normal processing (barrier=0) */
928         atomic_dec(&d->opd_sync_barrier);
929         osp_sync_check_for_work(d);
930
931         CDEBUG(D_CACHE, "%s: done in %lld: rc = %d\n", d->opd_obd->obd_name,
932                ktime_get_seconds() - start, rc);
933
934         RETURN(rc);
935 }
936
937 const struct dt_device_operations osp_dt_ops = {
938         .dt_statfs       = osp_statfs,
939         .dt_sync         = osp_sync,
940         .dt_trans_create = osp_trans_create,
941         .dt_trans_start  = osp_trans_start,
942         .dt_trans_stop   = osp_trans_stop,
943         .dt_trans_cb_add   = osp_trans_cb_add,
944 };
945
946 /**
947  * Connect OSP to local OSD.
948  *
949  * Locate the local OSD referenced by \a nextdev and connect to it. Sometimes,
950  * OSP needs to access the local OSD to store some information. For example,
951  * during precreate, it needs to update last used OID and sequence file
952  * (LAST_SEQ) in local OSD.
953  *
954  * \param[in] env       execution environment
955  * \param[in] osp       OSP device
956  * \param[in] nextdev   the name of local OSD
957  *
958  * \retval 0            0 connection succeeded
959  * \retval negative     negative errno connection failed
960  */
961 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *osp,
962                               const char *nextdev)
963 {
964         struct obd_connect_data *data = NULL;
965         struct obd_device       *obd;
966         int                      rc;
967
968         ENTRY;
969
970         LASSERT(osp->opd_storage_exp == NULL);
971
972         OBD_ALLOC_PTR(data);
973         if (data == NULL)
974                 RETURN(-ENOMEM);
975
976         obd = class_name2obd(nextdev);
977         if (obd == NULL) {
978                 CERROR("%s: can't locate next device: %s\n",
979                        osp->opd_obd->obd_name, nextdev);
980                 GOTO(out, rc = -ENOTCONN);
981         }
982
983         rc = obd_connect(env, &osp->opd_storage_exp, obd, &obd->obd_uuid, data,
984                          NULL);
985         if (rc) {
986                 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
987                        osp->opd_obd->obd_name, nextdev, rc);
988                 GOTO(out, rc);
989         }
990
991         osp->opd_dt_dev.dd_lu_dev.ld_site =
992                 osp->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
993         LASSERT(osp->opd_dt_dev.dd_lu_dev.ld_site);
994         osp->opd_storage = lu2dt_dev(osp->opd_storage_exp->exp_obd->obd_lu_dev);
995
996 out:
997         OBD_FREE_PTR(data);
998         RETURN(rc);
999 }
1000
1001 /**
1002  * Determine if the lock needs to be cancelled
1003  *
1004  * Determine if the unused lock should be cancelled before replay, see
1005  * (ldlm_cancel_no_wait_policy()). Currently, only inode bits lock exists
1006  * between MDTs.
1007  *
1008  * \param[in] lock      lock to be checked.
1009  *
1010  * \retval              1 if the lock needs to be cancelled before replay.
1011  * \retval              0 if the lock does not need to be cancelled before
1012  *                      replay.
1013  */
1014 static int osp_cancel_weight(struct ldlm_lock *lock)
1015 {
1016         if (lock->l_resource->lr_type != LDLM_IBITS)
1017                 RETURN(0);
1018
1019         RETURN(1);
1020 }
1021
1022 /**
1023  * Initialize OSP device according to the parameters in the configuration
1024  * log \a cfg.
1025  *
1026  * Reconstruct the local device name from the configuration profile, and
1027  * initialize necessary threads and structures according to the OSP type
1028  * (MDT or OST).
1029  *
1030  * Since there is no record in the MDT configuration for the local disk
1031  * device, we have to extract this from elsewhere in the profile.
1032  * The only information we get at setup is from the OSC records:
1033  * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
1034  *
1035  * Note: configs generated by Lustre 1.8 are missing the -MDTxxxx part,
1036  * so, we need to reconstruct the name of the underlying OSD from this:
1037  * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".
1038  *
1039  * \param[in] env       execution environment
1040  * \param[in] osp       OSP device
1041  * \param[in] ldt       lu device type of OSP
1042  * \param[in] cfg       configuration log
1043  *
1044  * \retval 0            0 if OSP initialization succeeded.
1045  * \retval negative     negative errno if OSP initialization failed.
1046  */
1047 static int osp_init0(const struct lu_env *env, struct osp_device *osp,
1048                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
1049 {
1050         struct obd_device       *obd;
1051         struct obd_import       *imp;
1052         char *src, *tgt, *osdname = NULL;
1053         const char *mdt;
1054         int                     rc;
1055         u32 idx;
1056
1057         ENTRY;
1058
1059         mutex_init(&osp->opd_async_requests_mutex);
1060         INIT_LIST_HEAD(&osp->opd_async_updates);
1061         init_rwsem(&osp->opd_async_updates_rwsem);
1062         atomic_set(&osp->opd_async_updates_count, 0);
1063
1064         obd = class_name2obd(lustre_cfg_string(cfg, 0));
1065         if (obd == NULL) {
1066                 CERROR("Cannot find obd with name %s\n",
1067                        lustre_cfg_string(cfg, 0));
1068                 RETURN(-ENODEV);
1069         }
1070         osp->opd_obd = obd;
1071
1072         src = lustre_cfg_string(cfg, 0);
1073         if (src == NULL)
1074                 RETURN(-EINVAL);
1075
1076         tgt = strrchr(src, '-');
1077         if (tgt == NULL) {
1078                 CERROR("%s: invalid target name %s: rc = %d\n",
1079                        osp->opd_obd->obd_name, lustre_cfg_string(cfg, 0),
1080                        -EINVAL);
1081                 RETURN(-EINVAL);
1082         }
1083
1084         if (strncmp(tgt, "-osc", 4) == 0) {
1085                 /* Old OSC name fsname-OSTXXXX-osc */
1086                 for (tgt--; tgt > src && *tgt != '-'; tgt--)
1087                         ;
1088                 if (tgt == src) {
1089                         CERROR("%s: invalid target name %s: rc = %d\n",
1090                                osp->opd_obd->obd_name,
1091                                lustre_cfg_string(cfg, 0), -EINVAL);
1092                         RETURN(-EINVAL);
1093                 }
1094
1095                 if (strncmp(tgt, "-OST", 4) != 0) {
1096                         CERROR("%s: invalid target name %s: rc = %d\n",
1097                                osp->opd_obd->obd_name,
1098                                lustre_cfg_string(cfg, 0), -EINVAL);
1099                         RETURN(-EINVAL);
1100                 }
1101
1102                 rc = target_name2index(tgt + 1, &idx, &mdt);
1103                 if (rc < 0 || rc & LDD_F_SV_ALL || mdt[0] != '-') {
1104                         CERROR("%s: invalid OST index in '%s': rc = %d\n",
1105                                osp->opd_obd->obd_name, src, -EINVAL);
1106                         RETURN(-EINVAL);
1107                 }
1108                 osp->opd_index = idx;
1109                 osp->opd_group = 0;
1110                 idx = tgt - src;
1111         } else {
1112                 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
1113                 if (strncmp(tgt, "-MDT", 4) != 0 &&
1114                     strncmp(tgt, "-OST", 4) != 0) {
1115                         CERROR("%s: invalid target name %s: rc = %d\n",
1116                                osp->opd_obd->obd_name,
1117                                lustre_cfg_string(cfg, 0), -EINVAL);
1118                         RETURN(-EINVAL);
1119                 }
1120
1121                 rc = target_name2index(tgt + 1, &idx, &mdt);
1122                 if (rc < 0 || rc & LDD_F_SV_ALL || *mdt != '\0') {
1123                         CERROR("%s: invalid OST index in '%s': rc = %d\n",
1124                                osp->opd_obd->obd_name, src, -EINVAL);
1125                         RETURN(-EINVAL);
1126                 }
1127
1128                 /* Get MDT index from the name and set it to opd_group,
1129                  * which will be used by OSP to connect with OST */
1130                 osp->opd_group = idx;
1131                 if (tgt - src <= 12) {
1132                         CERROR("%s: invalid mdt index from %s: rc =%d\n",
1133                                osp->opd_obd->obd_name,
1134                                lustre_cfg_string(cfg, 0), -EINVAL);
1135                         RETURN(-EINVAL);
1136                 }
1137
1138                 if (strncmp(tgt - 12, "-MDT", 4) == 0)
1139                         osp->opd_connect_mdt = 1;
1140
1141                 rc = target_name2index(tgt - 11, &idx, &mdt);
1142                 if (rc < 0 || rc & LDD_F_SV_ALL || mdt[0] != '-') {
1143                         CERROR("%s: invalid OST index in '%s': rc =%d\n",
1144                                osp->opd_obd->obd_name, src, -EINVAL);
1145                         RETURN(-EINVAL);
1146                 }
1147
1148                 osp->opd_index = idx;
1149                 idx = tgt - src - 12;
1150         }
1151         /* check the fsname length, and after this everything else will fit */
1152         if (idx > MTI_NAME_MAXLEN) {
1153                 CERROR("%s: fsname too long in '%s': rc = %d\n",
1154                        osp->opd_obd->obd_name, src, -EINVAL);
1155                 RETURN(-EINVAL);
1156         }
1157
1158         OBD_ALLOC(osdname, MAX_OBD_NAME);
1159         if (osdname == NULL)
1160                 RETURN(-ENOMEM);
1161
1162         memcpy(osdname, src, idx); /* copy just the fsname part */
1163         osdname[idx] = '\0';
1164
1165         mdt = strstr(mdt, "-MDT");
1166         if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
1167                 strcat(osdname, "-MDT0000");
1168         else
1169                 strcat(osdname, mdt);
1170         strcat(osdname, "-osd");
1171         CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
1172
1173         osp_init_rpc_lock(osp);
1174
1175         osp->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
1176         osp->opd_dt_dev.dd_ops = &osp_dt_ops;
1177
1178         obd->obd_lu_dev = &osp->opd_dt_dev.dd_lu_dev;
1179
1180         rc = osp_connect_to_osd(env, osp, osdname);
1181         if (rc)
1182                 GOTO(out_fini, rc);
1183
1184         rc = ptlrpcd_addref();
1185         if (rc)
1186                 GOTO(out_disconnect, rc);
1187
1188         rc = client_obd_setup(obd, cfg);
1189         if (rc) {
1190                 CERROR("%s: can't setup obd: rc = %d\n", osp->opd_obd->obd_name,
1191                        rc);
1192                 GOTO(out_ref, rc);
1193         }
1194
1195         osp_tunables_init(osp);
1196
1197         rc = obd_fid_init(osp->opd_obd, NULL, osp->opd_connect_mdt ?
1198                           LUSTRE_SEQ_METADATA : LUSTRE_SEQ_DATA);
1199         if (rc) {
1200                 CERROR("%s: fid init error: rc = %d\n",
1201                        osp->opd_obd->obd_name, rc);
1202                 GOTO(out_proc, rc);
1203         }
1204
1205         if (!osp->opd_connect_mdt) {
1206                 /* Initialize last id from the storage - will be
1207                  * used in orphan cleanup. */
1208                 if (!osp->opd_storage->dd_rdonly) {
1209                         rc = osp_last_used_init(env, osp);
1210                         if (rc)
1211                                 GOTO(out_fid, rc);
1212                 }
1213
1214                 /* Initialize precreation thread, it handles new
1215                  * connections as well. */
1216                 rc = osp_init_precreate(osp);
1217                 if (rc)
1218                         GOTO(out_last_used, rc);
1219
1220                 /*
1221                  * Initialize synhronization mechanism taking
1222                  * care of propogating changes to OST in near
1223                  * transactional manner.
1224                  */
1225                 rc = osp_sync_init(env, osp);
1226                 if (rc < 0)
1227                         GOTO(out_precreat, rc);
1228         } else {
1229                 osp->opd_got_disconnected = 1;
1230                 rc = osp_update_init(osp);
1231                 if (rc != 0)
1232                         GOTO(out_fid, rc);
1233         }
1234
1235         rc = osp_init_statfs(osp);
1236         if (rc)
1237                 GOTO(out_precreat, rc);
1238
1239         ns_register_cancel(obd->obd_namespace, osp_cancel_weight);
1240
1241         /*
1242          * Initiate connect to OST
1243          */
1244         imp = obd->u.cli.cl_import;
1245
1246         rc = ptlrpc_init_import(imp);
1247         if (rc)
1248                 GOTO(out, rc);
1249         if (osdname)
1250                 OBD_FREE(osdname, MAX_OBD_NAME);
1251         RETURN(0);
1252
1253 out:
1254         if (!osp->opd_connect_mdt)
1255                 /* stop sync thread */
1256                 osp_sync_fini(osp);
1257 out_precreat:
1258         /* stop precreate thread */
1259         if (!osp->opd_connect_mdt)
1260                 osp_precreate_fini(osp);
1261         else
1262                 osp_update_fini(env, osp);
1263 out_last_used:
1264         if (!osp->opd_connect_mdt)
1265                 osp_last_used_fini(env, osp);
1266 out_fid:
1267         obd_fid_fini(osp->opd_obd);
1268 out_proc:
1269         osp_tunables_fini(osp);
1270         client_obd_cleanup(obd);
1271 out_ref:
1272         ptlrpcd_decref();
1273 out_disconnect:
1274         obd_disconnect(osp->opd_storage_exp);
1275 out_fini:
1276         if (osdname)
1277                 OBD_FREE(osdname, MAX_OBD_NAME);
1278         RETURN(rc);
1279 }
1280
1281 /**
1282  * Implementation of lu_device_type_operations::ldto_device_free
1283  *
1284  * Free the OSP device in memory.  No return value is needed for now,
1285  * so always return NULL to comply with the interface.
1286  *
1287  * \param[in] env       execution environment
1288  * \param[in] lu        lu_device of OSP
1289  *
1290  * \retval NULL         NULL unconditionally
1291  */
1292 static struct lu_device *osp_device_free(const struct lu_env *env,
1293                                          struct lu_device *lu)
1294 {
1295         struct osp_device *osp = lu2osp_dev(lu);
1296
1297         if (atomic_read(&lu->ld_ref) && lu->ld_site) {
1298                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1299                 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
1300         }
1301         dt_device_fini(&osp->opd_dt_dev);
1302         OBD_FREE_PTR(osp);
1303
1304         return NULL;
1305 }
1306
1307 /**
1308  * Implementation of lu_device_type_operations::ldto_device_alloc
1309  *
1310  * This function allocates and initializes OSP device in memory according to
1311  * the config log.
1312  *
1313  * \param[in] env       execution environment
1314  * \param[in] type      device type of OSP
1315  * \param[in] lcfg      config log
1316  *
1317  * \retval pointer              the pointer of allocated OSP if succeed.
1318  * \retval ERR_PTR(errno)       ERR_PTR(errno) if failed.
1319  */
1320 static struct lu_device *osp_device_alloc(const struct lu_env *env,
1321                                           struct lu_device_type *type,
1322                                           struct lustre_cfg *lcfg)
1323 {
1324         struct osp_device *osp;
1325         struct lu_device  *ld;
1326
1327         OBD_ALLOC_PTR(osp);
1328         if (osp == NULL) {
1329                 ld = ERR_PTR(-ENOMEM);
1330         } else {
1331                 int rc;
1332
1333                 ld = osp2lu_dev(osp);
1334                 dt_device_init(&osp->opd_dt_dev, type);
1335                 rc = osp_init0(env, osp, type, lcfg);
1336                 if (rc != 0) {
1337                         osp_device_free(env, ld);
1338                         ld = ERR_PTR(rc);
1339                 }
1340         }
1341         return ld;
1342 }
1343
1344 /**
1345  * Implementation of lu_device_type_operations::ldto_device_fini
1346  *
1347  * This function cleans up the OSP device, i.e. release and free those
1348  * attached items in osp_device.
1349  *
1350  * \param[in] env       execution environment
1351  * \param[in] ld        lu_device of OSP
1352  *
1353  * \retval NULL                 NULL if cleanup succeeded.
1354  * \retval ERR_PTR(errno)       ERR_PTR(errno) if cleanup failed.
1355  */
1356 static struct lu_device *osp_device_fini(const struct lu_env *env,
1357                                          struct lu_device *ld)
1358 {
1359         struct osp_device *osp = lu2osp_dev(ld);
1360         int                rc;
1361
1362         ENTRY;
1363
1364         if (osp->opd_async_requests != NULL) {
1365                 osp_update_request_destroy(env, osp->opd_async_requests);
1366                 osp->opd_async_requests = NULL;
1367         }
1368
1369         if (osp->opd_storage_exp) {
1370                 /* wait for the commit callbacks to complete */
1371                 wait_event(osp->opd_sync_waitq,
1372                           atomic_read(&osp->opd_commits_registered) == 0);
1373                 obd_disconnect(osp->opd_storage_exp);
1374         }
1375
1376         LASSERT(osp->opd_obd);
1377
1378         rc = client_obd_cleanup(osp->opd_obd);
1379         if (rc != 0) {
1380                 ptlrpcd_decref();
1381                 RETURN(ERR_PTR(rc));
1382         }
1383
1384         osp_tunables_fini(osp);
1385
1386         ptlrpcd_decref();
1387
1388         RETURN(NULL);
1389 }
1390
1391 /**
1392  * Implementation of obd_ops::o_reconnect
1393  *
1394  * This function is empty and does not need to do anything for now.
1395  */
1396 static int osp_reconnect(const struct lu_env *env,
1397                          struct obd_export *exp, struct obd_device *obd,
1398                          struct obd_uuid *cluuid,
1399                          struct obd_connect_data *data,
1400                          void *localdata)
1401 {
1402         return 0;
1403 }
1404
1405 /*
1406  * Implementation of obd_ops::o_connect
1407  *
1408  * Connect OSP to the remote target (MDT or OST). Allocate the
1409  * export and return it to the LOD, which calls this function
1410  * for each OSP to connect it to the remote target. This function
1411  * is currently only called once per OSP.
1412  *
1413  * \param[in] env       execution environment
1414  * \param[out] exp      export connected to OSP
1415  * \param[in] obd       OSP device
1416  * \param[in] cluuid    OSP device client uuid
1417  * \param[in] data      connect_data to be used to connect to the remote
1418  *                      target
1419  * \param[in] localdata necessary for the API interface, but not used in
1420  *                      this function
1421  *
1422  * \retval 0            0 if the connection succeeded.
1423  * \retval negative     negative errno if the connection failed.
1424  */
1425 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
1426                            struct obd_device *obd, struct obd_uuid *cluuid,
1427                            struct obd_connect_data *data, void *localdata)
1428 {
1429         struct osp_device       *osp = lu2osp_dev(obd->obd_lu_dev);
1430         struct obd_connect_data *ocd;
1431         struct obd_import       *imp;
1432         struct lustre_handle     conn;
1433         int                      rc;
1434
1435         ENTRY;
1436
1437         CDEBUG(D_CONFIG, "connect #%d\n", osp->opd_connects);
1438
1439         rc = class_connect(&conn, obd, cluuid);
1440         if (rc)
1441                 RETURN(rc);
1442
1443         *exp = class_conn2export(&conn);
1444         /* Why should there ever be more than 1 connect? */
1445         osp->opd_connects++;
1446         LASSERT(osp->opd_connects == 1);
1447
1448         osp->opd_exp = *exp;
1449
1450         imp = osp->opd_obd->u.cli.cl_import;
1451         imp->imp_dlm_handle = conn;
1452
1453         LASSERT(data != NULL);
1454         LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
1455         ocd = &imp->imp_connect_data;
1456         *ocd = *data;
1457
1458         imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
1459         imp->imp_connect_flags2_orig = ocd->ocd_connect_flags2;
1460
1461         ocd->ocd_version = LUSTRE_VERSION_CODE;
1462         ocd->ocd_index = data->ocd_index;
1463
1464         rc = ptlrpc_connect_import(imp);
1465         if (rc) {
1466                 CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
1467                 GOTO(out, rc);
1468         } else {
1469                 osp->opd_obd->u.cli.cl_seq->lcs_exp =
1470                                 class_export_get(osp->opd_exp);
1471         }
1472
1473         ptlrpc_pinger_add_import(imp);
1474 out:
1475         RETURN(rc);
1476 }
1477
1478 /**
1479  * Implementation of obd_ops::o_disconnect
1480  *
1481  * Disconnect the export for the OSP.  This is called by LOD to release the
1482  * OSP during cleanup (\see lod_del_device()). The OSP will be released after
1483  * the export is released.
1484  *
1485  * \param[in] exp       export to be disconnected.
1486  *
1487  * \retval 0            0 if disconnection succeed
1488  * \retval negative     negative errno if disconnection failed
1489  */
1490 static int osp_obd_disconnect(struct obd_export *exp)
1491 {
1492         struct obd_device *obd = exp->exp_obd;
1493         struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
1494         int                rc;
1495         ENTRY;
1496
1497         /* Only disconnect the underlying layers on the final disconnect. */
1498         LASSERT(osp->opd_connects == 1);
1499         osp->opd_connects--;
1500
1501         rc = class_disconnect(exp);
1502         if (rc) {
1503                 CERROR("%s: class disconnect error: rc = %d\n",
1504                        obd->obd_name, rc);
1505                 RETURN(rc);
1506         }
1507
1508         /* destroy the device */
1509         class_manual_cleanup(obd);
1510
1511         RETURN(rc);
1512 }
1513
1514 /**
1515  * Implementation of obd_ops::o_statfs
1516  *
1517  * Send a RPC to the remote target to get statfs status. This is only used
1518  * in lprocfs helpers by obd_statfs.
1519  *
1520  * \param[in] env       execution environment
1521  * \param[in] exp       connection state from this OSP to the parent (LOD)
1522  *                      device
1523  * \param[out] osfs     hold the statfs result
1524  * \param[in] unused    Not used in this function for now
1525  * \param[in] flags     flags to indicate how OSP will issue the RPC
1526  *
1527  * \retval 0            0 if statfs succeeded.
1528  * \retval negative     negative errno if statfs failed.
1529  */
1530 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
1531                           struct obd_statfs *osfs, time64_t unused, __u32 flags)
1532 {
1533         struct obd_statfs       *msfs;
1534         struct ptlrpc_request   *req;
1535         struct obd_import       *imp = NULL;
1536         int                      rc;
1537
1538         ENTRY;
1539
1540         /* Since the request might also come from lprocfs, so we need
1541          * sync this with client_disconnect_export Bug15684 */
1542         down_read(&exp->exp_obd->u.cli.cl_sem);
1543         if (exp->exp_obd->u.cli.cl_import)
1544                 imp = class_import_get(exp->exp_obd->u.cli.cl_import);
1545         up_read(&exp->exp_obd->u.cli.cl_sem);
1546         if (!imp)
1547                 RETURN(-ENODEV);
1548
1549         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
1550
1551         class_import_put(imp);
1552
1553         if (req == NULL)
1554                 RETURN(-ENOMEM);
1555
1556         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
1557         if (rc) {
1558                 ptlrpc_request_free(req);
1559                 RETURN(rc);
1560         }
1561         ptlrpc_request_set_replen(req);
1562         req->rq_request_portal = OST_CREATE_PORTAL;
1563         ptlrpc_at_set_req_timeout(req);
1564
1565         if (flags & OBD_STATFS_NODELAY) {
1566                 /* procfs requests not want stat in wait for avoid deadlock */
1567                 req->rq_no_resend = 1;
1568                 req->rq_no_delay = 1;
1569         }
1570
1571         rc = ptlrpc_queue_wait(req);
1572         if (rc)
1573                 GOTO(out, rc);
1574
1575         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1576         if (msfs == NULL)
1577                 GOTO(out, rc = -EPROTO);
1578
1579         *osfs = *msfs;
1580
1581         EXIT;
1582 out:
1583         ptlrpc_req_finished(req);
1584         return rc;
1585 }
1586
1587 /**
1588  * Implementation of obd_ops::o_import_event
1589  *
1590  * This function is called when some related import event happens. It will
1591  * mark the necessary flags according to the event and notify the necessary
1592  * threads (mainly precreate thread).
1593  *
1594  * \param[in] obd       OSP OBD device
1595  * \param[in] imp       import attached from OSP to remote (OST/MDT) service
1596  * \param[in] event     event related to remote service (IMP_EVENT_*)
1597  *
1598  * \retval 0            0 if the event handling succeeded.
1599  * \retval negative     negative errno if the event handling failed.
1600  */
1601 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1602                             enum obd_import_event event)
1603 {
1604         struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1605         int rc;
1606
1607         switch (event) {
1608         case IMP_EVENT_DISCON:
1609                 d->opd_got_disconnected = 1;
1610                 d->opd_imp_connected = 0;
1611                 if (d->opd_connect_mdt)
1612                         break;
1613
1614                 if (d->opd_pre != NULL) {
1615                         osp_pre_update_status(d, -ENODEV);
1616                         wake_up(&d->opd_pre_waitq);
1617                 }
1618
1619                 CDEBUG(D_HA, "got disconnected\n");
1620                 break;
1621         case IMP_EVENT_INACTIVE:
1622                 d->opd_imp_active = 0;
1623                 d->opd_imp_connected = 0;
1624                 d->opd_obd->obd_inactive = 1;
1625                 if (d->opd_connect_mdt)
1626                         break;
1627                 if (d->opd_pre != NULL) {
1628                         /* Import is invalid, we can`t get stripes so
1629                          * wakeup waiters */
1630                         rc = imp->imp_deactive ? -ESHUTDOWN : -ENODEV;
1631                         osp_pre_update_status(d, rc);
1632                         wake_up(&d->opd_pre_waitq);
1633                 }
1634
1635                 CDEBUG(D_HA, "got inactive\n");
1636                 break;
1637         case IMP_EVENT_ACTIVE:
1638                 d->opd_imp_active = 1;
1639
1640                 if (d->opd_got_disconnected)
1641                         d->opd_new_connection = 1;
1642                 d->opd_imp_connected = 1;
1643                 d->opd_imp_seen_connected = 1;
1644                 d->opd_obd->obd_inactive = 0;
1645                 wake_up(&d->opd_pre_waitq);
1646                 if (d->opd_connect_mdt)
1647                         break;
1648
1649                 osp_sync_check_for_work(d);
1650                 CDEBUG(D_HA, "got connected\n");
1651                 break;
1652         case IMP_EVENT_INVALIDATE:
1653                 if (d->opd_connect_mdt)
1654                         osp_invalidate_request(d);
1655
1656                 if (obd->obd_namespace == NULL)
1657                         break;
1658                 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1659                 break;
1660         case IMP_EVENT_OCD:
1661         case IMP_EVENT_DEACTIVATE:
1662         case IMP_EVENT_ACTIVATE:
1663                 break;
1664         default:
1665                 CERROR("%s: unsupported import event: %#x\n",
1666                        obd->obd_name, event);
1667         }
1668         return 0;
1669 }
1670
1671 /**
1672  * Implementation of obd_ops: o_iocontrol
1673  *
1674  * This function is the ioctl handler for OSP. Note: lctl will access the OSP
1675  * directly by ioctl, instead of through the MDS stack.
1676  *
1677  * param[in] cmd        ioctl command.
1678  * param[in] exp        export of this OSP.
1679  * param[in] len        data length of \a karg.
1680  * param[in] karg       input argument which is packed as
1681  *                      obd_ioctl_data
1682  * param[out] uarg      pointer to userspace buffer (must access by
1683  *                      copy_to_user()).
1684  *
1685  * \retval 0            0 if the ioctl handling succeeded.
1686  * \retval negative     negative errno if the ioctl handling failed.
1687  */
1688 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1689                          void *karg, void __user *uarg)
1690 {
1691         struct obd_device       *obd = exp->exp_obd;
1692         struct osp_device       *d;
1693         struct obd_ioctl_data   *data = karg;
1694         int                      rc = 0;
1695
1696         ENTRY;
1697
1698         LASSERT(obd->obd_lu_dev);
1699         d = lu2osp_dev(obd->obd_lu_dev);
1700         LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1701
1702         if (!try_module_get(THIS_MODULE)) {
1703                 CERROR("%s: cannot get module '%s'\n", obd->obd_name,
1704                        module_name(THIS_MODULE));
1705                 return -EINVAL;
1706         }
1707
1708         switch (cmd) {
1709         case OBD_IOC_CLIENT_RECOVER:
1710                 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1711                                            data->ioc_inlbuf1, 0);
1712                 if (rc > 0)
1713                         rc = 0;
1714                 break;
1715         case IOC_OSC_SET_ACTIVE:
1716                 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1717                                               data->ioc_offset);
1718                 break;
1719         default:
1720                 CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
1721                        cmd, current->comm);
1722                 rc = -ENOTTY;
1723         }
1724         module_put(THIS_MODULE);
1725         return rc;
1726 }
1727
1728
1729 /**
1730  * Implementation of obd_ops::o_get_info
1731  *
1732  * Retrieve information by key. Retrieval starts from the top layer
1733  * (MDT) of the MDS stack and traverses the stack by calling the
1734  * obd_get_info() method of the next sub-layer.
1735  *
1736  * \param[in] env       execution environment
1737  * \param[in] exp       export of this OSP
1738  * \param[in] keylen    length of \a key
1739  * \param[in] key       the key
1740  * \param[out] vallen   length of \a val
1741  * \param[out] val      holds the value returned by the key
1742  *
1743  * \retval 0            0 if getting information succeeded.
1744  * \retval negative     negative errno if getting information failed.
1745  */
1746 static int osp_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1747                             __u32 keylen, void *key, __u32 *vallen, void *val)
1748 {
1749         int rc = -EINVAL;
1750
1751         if (KEY_IS(KEY_OSP_CONNECTED)) {
1752                 struct obd_device       *obd = exp->exp_obd;
1753                 struct osp_device       *osp;
1754
1755                 if (!obd->obd_set_up || obd->obd_stopping)
1756                         RETURN(-EAGAIN);
1757
1758                 osp = lu2osp_dev(obd->obd_lu_dev);
1759                 LASSERT(osp);
1760                 /*
1761                  * 1.8/2.0 behaviour is that OST being connected once at least
1762                  * is considered "healthy". and one "healthy" OST is enough to
1763                  * allow lustre clients to connect to MDS
1764                  */
1765                 RETURN(!osp->opd_imp_seen_connected);
1766         }
1767
1768         RETURN(rc);
1769 }
1770
1771 static int osp_obd_set_info_async(const struct lu_env *env,
1772                                   struct obd_export *exp,
1773                                   u32 keylen, void *key,
1774                                   u32 vallen, void *val,
1775                                   struct ptlrpc_request_set *set)
1776 {
1777         struct obd_device       *obd = exp->exp_obd;
1778         struct obd_import       *imp = obd->u.cli.cl_import;
1779         struct osp_device       *osp;
1780         struct ptlrpc_request   *req;
1781         char                    *tmp;
1782         int                      rc;
1783
1784         if (KEY_IS(KEY_SPTLRPC_CONF)) {
1785                 sptlrpc_conf_client_adapt(exp->exp_obd);
1786                 RETURN(0);
1787         }
1788
1789         LASSERT(set != NULL);
1790         if (!obd->obd_set_up || obd->obd_stopping)
1791                 RETURN(-EAGAIN);
1792         osp = lu2osp_dev(obd->obd_lu_dev);
1793
1794         req = ptlrpc_request_alloc(imp, &RQF_OBD_SET_INFO);
1795         if (req == NULL)
1796                 RETURN(-ENOMEM);
1797
1798         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_KEY,
1799                              RCL_CLIENT, keylen);
1800         req_capsule_set_size(&req->rq_pill, &RMF_SETINFO_VAL,
1801                              RCL_CLIENT, vallen);
1802         if (osp->opd_connect_mdt)
1803                 rc = ptlrpc_request_pack(req, LUSTRE_MDS_VERSION, MDS_SET_INFO);
1804         else
1805                 rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_SET_INFO);
1806         if (rc) {
1807                 ptlrpc_request_free(req);
1808                 RETURN(rc);
1809         }
1810
1811         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_KEY);
1812         memcpy(tmp, key, keylen);
1813         tmp = req_capsule_client_get(&req->rq_pill, &RMF_SETINFO_VAL);
1814         memcpy(tmp, val, vallen);
1815
1816         ptlrpc_request_set_replen(req);
1817         ptlrpc_set_add_req(set, req);
1818         ptlrpc_check_set(NULL, set);
1819
1820         RETURN(0);
1821 }
1822
1823 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1824 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1825 static void osp_key_exit(const struct lu_context *ctx,
1826                          struct lu_context_key *key, void *data)
1827 {
1828         struct osp_thread_info *info = data;
1829
1830         info->osi_attr.la_valid = 0;
1831 }
1832
1833 struct lu_context_key osp_thread_key = {
1834         .lct_tags = LCT_MD_THREAD,
1835         .lct_init = osp_key_init,
1836         .lct_fini = osp_key_fini,
1837         .lct_exit = osp_key_exit
1838 };
1839
1840 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1841 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1842
1843 struct lu_context_key osp_txn_key = {
1844         .lct_tags = LCT_OSP_THREAD,
1845         .lct_init = osp_txn_key_init,
1846         .lct_fini = osp_txn_key_fini
1847 };
1848 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1849
1850 static struct lu_device_type_operations osp_device_type_ops = {
1851         .ldto_init           = osp_type_init,
1852         .ldto_fini           = osp_type_fini,
1853
1854         .ldto_start          = osp_type_start,
1855         .ldto_stop           = osp_type_stop,
1856
1857         .ldto_device_alloc   = osp_device_alloc,
1858         .ldto_device_free    = osp_device_free,
1859
1860         .ldto_device_fini    = osp_device_fini
1861 };
1862
1863 static struct lu_device_type osp_device_type = {
1864         .ldt_tags     = LU_DEVICE_DT,
1865         .ldt_name     = LUSTRE_OSP_NAME,
1866         .ldt_ops      = &osp_device_type_ops,
1867         .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD,
1868 };
1869
1870 static const struct obd_ops osp_obd_device_ops = {
1871         .o_owner        = THIS_MODULE,
1872         .o_add_conn     = client_import_add_conn,
1873         .o_del_conn     = client_import_del_conn,
1874         .o_reconnect    = osp_reconnect,
1875         .o_connect      = osp_obd_connect,
1876         .o_disconnect   = osp_obd_disconnect,
1877         .o_get_info     = osp_obd_get_info,
1878         .o_set_info_async = osp_obd_set_info_async,
1879         .o_import_event = osp_import_event,
1880         .o_iocontrol    = osp_iocontrol,
1881         .o_statfs       = osp_obd_statfs,
1882         .o_fid_init     = client_fid_init,
1883         .o_fid_fini     = client_fid_fini,
1884 };
1885
1886 /**
1887  * Initialize OSP module.
1888  *
1889  * Register device types OSP and Light Weight Proxy (LWP) (\see lwp_dev.c)
1890  * in obd_types (\see class_obd.c).  Initialize procfs for the
1891  * the OSP device.  Note: OSP was called OSC before Lustre 2.4,
1892  * so for compatibility it still uses the name "osc" in procfs.
1893  * This is called at module load time.
1894  *
1895  * \retval 0            0 if initialization succeeds.
1896  * \retval negative     negative errno if initialization failed.
1897  */
1898 static int __init osp_init(void)
1899 {
1900         struct obd_type *sym;
1901         int rc;
1902
1903         rc = lu_kmem_init(osp_caches);
1904         if (rc)
1905                 return rc;
1906
1907         rc = class_register_type(&osp_obd_device_ops, NULL, false, NULL,
1908                                  LUSTRE_OSP_NAME, &osp_device_type);
1909         if (rc != 0) {
1910                 lu_kmem_fini(osp_caches);
1911                 return rc;
1912         }
1913
1914         rc = class_register_type(&lwp_obd_device_ops, NULL, false, NULL,
1915                                  LUSTRE_LWP_NAME, &lwp_device_type);
1916         if (rc != 0) {
1917                 class_unregister_type(LUSTRE_OSP_NAME);
1918                 lu_kmem_fini(osp_caches);
1919                 return rc;
1920         }
1921
1922         /* create "osc" entry for compatibility purposes */
1923         sym = class_add_symlinks(LUSTRE_OSC_NAME, true);
1924         if (IS_ERR(sym)) {
1925                 rc = PTR_ERR(sym);
1926                 /* does real "osc" already exist ? */
1927                 if (rc == -EEXIST)
1928                         rc = 0;
1929         }
1930
1931         return rc;
1932 }
1933
1934 /**
1935  * Finalize OSP module.
1936  *
1937  * This callback is called when kernel unloads OSP module from memory, and
1938  * it will deregister OSP and LWP device type from obd_types (\see class_obd.c).
1939  */
1940 static void __exit osp_exit(void)
1941 {
1942         struct obd_type *sym = class_search_type(LUSTRE_OSC_NAME);
1943
1944         /* if this was never fully initialized by the osc layer
1945          * then we are responsible for freeing this obd_type
1946          */
1947         if (sym) {
1948                 /* final put if we manage this obd type */
1949                 if (sym->typ_sym_filter)
1950                         kobject_put(&sym->typ_kobj);
1951                 /* put reference taken by class_search_type */
1952                 kobject_put(&sym->typ_kobj);
1953         }
1954
1955         class_unregister_type(LUSTRE_LWP_NAME);
1956         class_unregister_type(LUSTRE_OSP_NAME);
1957         lu_kmem_fini(osp_caches);
1958 }
1959
1960 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
1961 MODULE_DESCRIPTION("Lustre OSD Storage Proxy ("LUSTRE_OSP_NAME")");
1962 MODULE_VERSION(LUSTRE_VERSION_STRING);
1963 MODULE_LICENSE("GPL");
1964
1965 module_init(osp_init);
1966 module_exit(osp_exit);