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