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