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