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