Whamcloud - gitweb
52ad482ebd28567553c4f3b2325baaabd43a18d0
[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, 2013, 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  * Lustre OST Proxy Device
39  *
40  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
41  * Author: Mikhail Pershin <mike.pershin@intel.com>
42  * Author: Di Wang <di.wang@intel.com>
43  */
44
45 #define DEBUG_SUBSYSTEM S_MDS
46
47 #include <obd_class.h>
48 #include <lustre_param.h>
49 #include <lustre_log.h>
50 #include <lustre_mdc.h>
51
52 #include "osp_internal.h"
53
54 /* Slab for OSP object allocation */
55 struct kmem_cache *osp_object_kmem;
56
57 static struct lu_kmem_descr osp_caches[] = {
58         {
59                 .ckd_cache = &osp_object_kmem,
60                 .ckd_name  = "osp_obj",
61                 .ckd_size  = sizeof(struct osp_object)
62         },
63         {
64                 .ckd_cache = NULL
65         }
66 };
67
68 struct lu_object *osp_object_alloc(const struct lu_env *env,
69                                    const struct lu_object_header *hdr,
70                                    struct lu_device *d)
71 {
72         struct lu_object_header *h = NULL;
73         struct osp_object       *o;
74         struct lu_object        *l;
75
76         OBD_SLAB_ALLOC_PTR_GFP(o, osp_object_kmem, __GFP_IO);
77         if (o != NULL) {
78                 l = &o->opo_obj.do_lu;
79
80                 /* For data object, OSP obj would always be the top
81                  * object, i.e. hdr is always NULL, see lu_object_alloc.
82                  * But for metadata object, we always build the object
83                  * stack from MDT. i.e. mdt_object will be the top object
84                  * i.e.  hdr != NULL */
85                 if (hdr == NULL) {
86                         /* object for OST */
87                         h = &o->opo_header;
88                         lu_object_header_init(h);
89                         dt_object_init(&o->opo_obj, h, d);
90                         lu_object_add_top(h, l);
91                 } else {
92                         dt_object_init(&o->opo_obj, h, d);
93                 }
94
95                 l->lo_ops = &osp_lu_obj_ops;
96
97                 return l;
98         } else {
99                 return NULL;
100         }
101 }
102
103 static struct dt_object
104 *osp_find_or_create(const struct lu_env *env, struct osp_device *osp,
105                     struct lu_attr *attr, __u32 reg_id)
106 {
107         struct osp_thread_info *osi = osp_env_info(env);
108         struct dt_object_format dof = { 0 };
109         struct dt_object       *dto;
110         int                  rc;
111         ENTRY;
112
113         lu_local_obj_fid(&osi->osi_fid, reg_id);
114         attr->la_valid = LA_MODE;
115         attr->la_mode = S_IFREG | 0644;
116         dof.dof_type = DFT_REGULAR;
117         dto = dt_find_or_create(env, osp->opd_storage, &osi->osi_fid,
118                                 &dof, attr);
119         if (IS_ERR(dto))
120                 RETURN(dto);
121
122         rc = dt_attr_get(env, dto, attr, NULL);
123         if (rc) {
124                 CERROR("%s: can't be initialized: rc = %d\n",
125                        osp->opd_obd->obd_name, rc);
126                 lu_object_put(env, &dto->do_lu);
127                 RETURN(ERR_PTR(rc));
128         }
129         RETURN(dto);
130 }
131
132 static int osp_write_local_file(const struct lu_env *env,
133                                 struct osp_device *osp,
134                                 struct dt_object *dt_obj,
135                                 struct lu_buf *buf,
136                                 loff_t offset)
137 {
138         struct thandle *th;
139         int rc;
140
141         th = dt_trans_create(env, osp->opd_storage);
142         if (IS_ERR(th))
143                 RETURN(PTR_ERR(th));
144
145         rc = dt_declare_record_write(env, dt_obj, buf->lb_len, offset, th);
146         if (rc)
147                 GOTO(out, rc);
148         rc = dt_trans_start_local(env, osp->opd_storage, th);
149         if (rc)
150                 GOTO(out, rc);
151
152         rc = dt_record_write(env, dt_obj, buf, &offset, th);
153 out:
154         dt_trans_stop(env, osp->opd_storage, th);
155         RETURN(rc);
156 }
157
158 static int osp_init_last_objid(const struct lu_env *env, struct osp_device *osp)
159 {
160         struct osp_thread_info  *osi = osp_env_info(env);
161         struct lu_fid           *fid = &osp->opd_last_used_fid;
162         struct dt_object        *dto;
163         int                     rc;
164         ENTRY;
165
166         dto = osp_find_or_create(env, osp, &osi->osi_attr, MDD_LOV_OBJ_OID);
167         if (IS_ERR(dto))
168                 RETURN(PTR_ERR(dto));
169         /* object will be released in device cleanup path */
170         if (osi->osi_attr.la_size >=
171             sizeof(osi->osi_id) * (osp->opd_index + 1)) {
172                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
173                                    osp->opd_index);
174                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
175                 if (rc != 0)
176                         GOTO(out, rc);
177         } else {
178                 fid->f_oid = 0;
179                 osp_objid_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_oid,
180                                    osp->opd_index);
181                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
182                                           osi->osi_off);
183         }
184         osp->opd_last_used_oid_file = dto;
185         RETURN(0);
186 out:
187         /* object will be released in device cleanup path */
188         CERROR("%s: can't initialize lov_objid: rc = %d\n",
189                osp->opd_obd->obd_name, rc);
190         lu_object_put(env, &dto->do_lu);
191         osp->opd_last_used_oid_file = NULL;
192         RETURN(rc);
193 }
194
195 static int osp_init_last_seq(const struct lu_env *env, struct osp_device *osp)
196 {
197         struct osp_thread_info  *osi = osp_env_info(env);
198         struct lu_fid           *fid = &osp->opd_last_used_fid;
199         struct dt_object        *dto;
200         int                     rc;
201         ENTRY;
202
203         dto = osp_find_or_create(env, osp, &osi->osi_attr, MDD_LOV_OBJ_OSEQ);
204         if (IS_ERR(dto))
205                 RETURN(PTR_ERR(dto));
206
207         /* object will be released in device cleanup path */
208         if (osi->osi_attr.la_size >=
209             sizeof(osi->osi_id) * (osp->opd_index + 1)) {
210                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
211                                    osp->opd_index);
212                 rc = dt_record_read(env, dto, &osi->osi_lb, &osi->osi_off);
213                 if (rc != 0)
214                         GOTO(out, rc);
215         } else {
216                 fid->f_seq = 0;
217                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off, &fid->f_seq,
218                                     osp->opd_index);
219                 rc = osp_write_local_file(env, osp, dto, &osi->osi_lb,
220                                           osi->osi_off);
221         }
222         osp->opd_last_used_seq_file = dto;
223         RETURN(0);
224 out:
225         /* object will be released in device cleanup path */
226         CERROR("%s: can't initialize lov_seq: rc = %d\n",
227                osp->opd_obd->obd_name, rc);
228         lu_object_put(env, &dto->do_lu);
229         osp->opd_last_used_seq_file = NULL;
230         RETURN(rc);
231 }
232
233 static int osp_last_used_init(const struct lu_env *env, struct osp_device *osp)
234 {
235         struct osp_thread_info *osi = osp_env_info(env);
236         int                  rc;
237         ENTRY;
238
239         fid_zero(&osp->opd_last_used_fid);
240         rc = osp_init_last_objid(env, osp);
241         if (rc < 0) {
242                 CERROR("%s: Can not get ids %d from old objid!\n",
243                        osp->opd_obd->obd_name, rc);
244                 RETURN(rc);
245         }
246
247         rc = osp_init_last_seq(env, osp);
248         if (rc < 0) {
249                 CERROR("%s: Can not get ids %d from old objid!\n",
250                        osp->opd_obd->obd_name, rc);
251                 GOTO(out, rc);
252         }
253
254         if (fid_oid(&osp->opd_last_used_fid) != 0 &&
255             fid_seq(&osp->opd_last_used_fid) == 0) {
256                 /* Just upgrade from the old version,
257                  * set the seq to be IDIF */
258                 osp->opd_last_used_fid.f_seq =
259                    fid_idif_seq(fid_oid(&osp->opd_last_used_fid),
260                                 osp->opd_index);
261                 osp_objseq_buf_prep(&osi->osi_lb, &osi->osi_off,
262                                     &osp->opd_last_used_fid.f_seq,
263                                     osp->opd_index);
264                 rc = osp_write_local_file(env, osp, osp->opd_last_used_seq_file,
265                                           &osi->osi_lb, osi->osi_off);
266                 if (rc) {
267                         CERROR("%s : Can not write seq file: rc = %d\n",
268                                osp->opd_obd->obd_name, rc);
269                         GOTO(out, rc);
270                 }
271         }
272
273         if (!fid_is_zero(&osp->opd_last_used_fid) &&
274                  !fid_is_sane(&osp->opd_last_used_fid)) {
275                 CERROR("%s: Got invalid FID "DFID"\n", osp->opd_obd->obd_name,
276                         PFID(&osp->opd_last_used_fid));
277                 GOTO(out, rc = -EINVAL);
278         }
279
280         CDEBUG(D_INFO, "%s: Init last used fid "DFID"\n",
281                osp->opd_obd->obd_name, PFID(&osp->opd_last_used_fid));
282 out:
283         if (rc != 0) {
284                 if (osp->opd_last_used_oid_file != NULL) {
285                         lu_object_put(env, &osp->opd_last_used_oid_file->do_lu);
286                         osp->opd_last_used_oid_file = NULL;
287                 }
288                 if (osp->opd_last_used_seq_file != NULL) {
289                         lu_object_put(env, &osp->opd_last_used_seq_file->do_lu);
290                         osp->opd_last_used_seq_file = NULL;
291                 }
292         }
293
294         RETURN(rc);
295 }
296
297 static void osp_last_used_fini(const struct lu_env *env, struct osp_device *d)
298 {
299         /* release last_used file */
300         if (d->opd_last_used_oid_file != NULL) {
301                 lu_object_put(env, &d->opd_last_used_oid_file->do_lu);
302                 d->opd_last_used_oid_file = NULL;
303         }
304
305         if (d->opd_last_used_seq_file != NULL) {
306                 lu_object_put(env, &d->opd_last_used_seq_file->do_lu);
307                 d->opd_last_used_seq_file = NULL;
308         }
309 }
310
311 static int osp_disconnect(struct osp_device *d)
312 {
313         struct obd_import *imp;
314         int rc = 0;
315
316         imp = d->opd_obd->u.cli.cl_import;
317
318         /* Mark import deactivated now, so we don't try to reconnect if any
319          * of the cleanup RPCs fails (e.g. ldlm cancel, etc).  We don't
320          * fully deactivate the import, or that would drop all requests. */
321         LASSERT(imp != NULL);
322         spin_lock(&imp->imp_lock);
323         imp->imp_deactive = 1;
324         spin_unlock(&imp->imp_lock);
325
326         ptlrpc_deactivate_import(imp);
327
328         /* Some non-replayable imports (MDS's OSCs) are pinged, so just
329          * delete it regardless.  (It's safe to delete an import that was
330          * never added.) */
331         (void)ptlrpc_pinger_del_import(imp);
332
333         rc = ptlrpc_disconnect_import(imp, 0);
334         if (rc == -ETIMEDOUT || rc == -ENOTCONN || rc == -ESHUTDOWN)
335                 rc = 0;
336         if (rc)
337                 CERROR("%s: can't disconnect: rc = %d\n",
338                        d->opd_obd->obd_name, rc);
339
340         ptlrpc_invalidate_import(imp);
341
342         RETURN(rc);
343 }
344
345 static int osp_shutdown(const struct lu_env *env, struct osp_device *d)
346 {
347         int                      rc = 0;
348         ENTRY;
349
350         LASSERT(env);
351
352         rc = osp_disconnect(d);
353
354         if (!d->opd_connect_mdt) {
355                 /* stop precreate thread */
356                 osp_precreate_fini(d);
357
358                 /* stop sync thread */
359                 osp_sync_fini(d);
360
361                 /* release last_used file */
362                 osp_last_used_fini(env, d);
363         }
364
365         obd_fid_fini(d->opd_obd);
366
367         RETURN(rc);
368 }
369
370 static int osp_process_config(const struct lu_env *env,
371                               struct lu_device *dev, struct lustre_cfg *lcfg)
372 {
373         struct osp_device               *d = lu2osp_dev(dev);
374         struct lprocfs_static_vars       lvars = { 0 };
375         int                              rc;
376
377         ENTRY;
378
379         switch (lcfg->lcfg_command) {
380         case LCFG_PRE_CLEANUP:
381                 rc = osp_disconnect(d);
382                 break;
383         case LCFG_CLEANUP:
384                 lu_dev_del_linkage(dev->ld_site, dev);
385                 rc = osp_shutdown(env, d);
386                 break;
387         case LCFG_PARAM:
388                 lprocfs_osp_init_vars(&lvars);
389
390                 LASSERT(d->opd_obd);
391                 rc = class_process_proc_param(PARAM_OSC, lvars.obd_vars,
392                                               lcfg, d->opd_obd);
393                 if (rc > 0)
394                         rc = 0;
395                 if (rc == -ENOSYS) {
396                         /* class_process_proc_param() haven't found matching
397                          * parameter and returned ENOSYS so that layer(s)
398                          * below could use that. But OSP is the bottom, so
399                          * just ignore it */
400                         CERROR("%s: unknown param %s\n",
401                                (char *)lustre_cfg_string(lcfg, 0),
402                                (char *)lustre_cfg_string(lcfg, 1));
403                         rc = 0;
404                 }
405                 break;
406         default:
407                 CERROR("%s: unknown command %u\n",
408                        (char *)lustre_cfg_string(lcfg, 0), lcfg->lcfg_command);
409                 rc = 0;
410                 break;
411         }
412
413         RETURN(rc);
414 }
415
416 static int osp_recovery_complete(const struct lu_env *env,
417                                  struct lu_device *dev)
418 {
419         struct osp_device       *osp = lu2osp_dev(dev);
420         int                      rc = 0;
421
422         ENTRY;
423         osp->opd_recovery_completed = 1;
424         if (!osp->opd_connect_mdt)
425                 cfs_waitq_signal(&osp->opd_pre_waitq);
426         RETURN(rc);
427 }
428
429 const struct lu_device_operations osp_lu_ops = {
430         .ldo_object_alloc       = osp_object_alloc,
431         .ldo_process_config     = osp_process_config,
432         .ldo_recovery_complete  = osp_recovery_complete,
433 };
434
435 /**
436  * provides with statfs from corresponded OST
437  *
438  */
439 static int osp_statfs(const struct lu_env *env, struct dt_device *dev,
440                       struct obd_statfs *sfs)
441 {
442         struct osp_device *d = dt2osp_dev(dev);
443
444         ENTRY;
445
446         if (unlikely(d->opd_imp_active == 0))
447                 RETURN(-ENOTCONN);
448
449         /* return recently updated data */
450         *sfs = d->opd_statfs;
451
452         /*
453          * layer above osp (usually lod) can use ffree to estimate
454          * how many objects are available for immediate creation
455          */
456
457         spin_lock(&d->opd_pre_lock);
458         LASSERTF(fid_seq(&d->opd_pre_last_created_fid) ==
459                  fid_seq(&d->opd_pre_used_fid),
460                  "last_created "DFID", next_fid "DFID"\n",
461                  PFID(&d->opd_pre_last_created_fid),
462                  PFID(&d->opd_pre_used_fid));
463         sfs->os_fprecreated = fid_oid(&d->opd_pre_last_created_fid) -
464                               fid_oid(&d->opd_pre_used_fid);
465         sfs->os_fprecreated -= d->opd_pre_reserved;
466         spin_unlock(&d->opd_pre_lock);
467
468         LASSERT(sfs->os_fprecreated <= OST_MAX_PRECREATE * 2);
469
470         CDEBUG(D_OTHER, "%s: "LPU64" blocks, "LPU64" free, "LPU64" avail, "
471                LPU64" files, "LPU64" free files\n", d->opd_obd->obd_name,
472                sfs->os_blocks, sfs->os_bfree, sfs->os_bavail,
473                sfs->os_files, sfs->os_ffree);
474         RETURN(0);
475 }
476
477 static int osp_sync(const struct lu_env *env, struct dt_device *dev)
478 {
479         ENTRY;
480
481         /*
482          * XXX: wake up sync thread, command it to start flushing asap?
483          */
484
485         RETURN(0);
486 }
487
488 const struct dt_device_operations osp_dt_ops = {
489         .dt_statfs      = osp_statfs,
490         .dt_sync        = osp_sync,
491         .dt_trans_start = osp_trans_start,
492         .dt_trans_stop  = osp_trans_stop,
493 };
494
495 static int osp_connect_to_osd(const struct lu_env *env, struct osp_device *m,
496                               const char *nextdev)
497 {
498         struct obd_connect_data *data = NULL;
499         struct obd_device       *obd;
500         int                      rc;
501
502         ENTRY;
503
504         LASSERT(m->opd_storage_exp == NULL);
505
506         OBD_ALLOC_PTR(data);
507         if (data == NULL)
508                 RETURN(-ENOMEM);
509
510         obd = class_name2obd(nextdev);
511         if (obd == NULL) {
512                 CERROR("%s: can't locate next device: %s\n",
513                        m->opd_obd->obd_name, nextdev);
514                 GOTO(out, rc = -ENOTCONN);
515         }
516
517         rc = obd_connect(env, &m->opd_storage_exp, obd, &obd->obd_uuid, data,
518                          NULL);
519         if (rc) {
520                 CERROR("%s: cannot connect to next dev %s: rc = %d\n",
521                        m->opd_obd->obd_name, nextdev, rc);
522                 GOTO(out, rc);
523         }
524
525         m->opd_dt_dev.dd_lu_dev.ld_site =
526                 m->opd_storage_exp->exp_obd->obd_lu_dev->ld_site;
527         LASSERT(m->opd_dt_dev.dd_lu_dev.ld_site);
528         m->opd_storage = lu2dt_dev(m->opd_storage_exp->exp_obd->obd_lu_dev);
529
530 out:
531         OBD_FREE_PTR(data);
532         RETURN(rc);
533 }
534
535 static int osp_init0(const struct lu_env *env, struct osp_device *m,
536                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
537 {
538         struct obd_device       *obd;
539         struct obd_import       *imp;
540         class_uuid_t            uuid;
541         char                    *src, *tgt, *mdt, *osdname = NULL;
542         int                     rc, idx;
543
544         ENTRY;
545
546         obd = class_name2obd(lustre_cfg_string(cfg, 0));
547         if (obd == NULL) {
548                 CERROR("Cannot find obd with name %s\n",
549                        lustre_cfg_string(cfg, 0));
550                 RETURN(-ENODEV);
551         }
552         m->opd_obd = obd;
553
554         /* There is no record in the MDT configuration for the local disk
555          * device, so we have to extract this from elsewhere in the profile.
556          * The only information we get at setup is from the OSC records:
557          * setup 0:{fsname}-OSTxxxx-osc[-MDTxxxx] 1:lustre-OST0000_UUID 2:NID
558          * Note that 1.8 generated configs are missing the -MDTxxxx part.
559          * We need to reconstruct the name of the underlying OSD from this:
560          * {fsname}-{svname}-osd, for example "lustre-MDT0000-osd".  We
561          * also need to determine the OST index from this - will be used
562          * to calculate the offset in shared lov_objids file later */
563
564         src = lustre_cfg_string(cfg, 0);
565         if (src == NULL)
566                 RETURN(-EINVAL);
567
568         tgt = strrchr(src, '-');
569         if (tgt == NULL) {
570                 CERROR("%s: invalid target name %s\n",
571                        m->opd_obd->obd_name, lustre_cfg_string(cfg, 0));
572                 RETURN(-EINVAL);
573         }
574
575         if (strncmp(tgt, "-osc", 4) == 0) {
576                 /* Old OSC name fsname-OSTXXXX-osc */
577                 for (tgt--; tgt > src && *tgt != '-'; tgt--)
578                         ;
579                 if (tgt == src) {
580                         CERROR("%s: invalid target name %s\n",
581                                m->opd_obd->obd_name, lustre_cfg_string(cfg, 0));
582                         RETURN(-EINVAL);
583                 }
584
585                 if (strncmp(tgt, "-OST", 4) != 0) {
586                         CERROR("%s: invalid target name %s\n",
587                                m->opd_obd->obd_name, lustre_cfg_string(cfg, 0));
588                         RETURN(-EINVAL);
589                 }
590
591                 idx = simple_strtol(tgt + 4, &mdt, 16);
592                 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
593                         CERROR("%s: invalid OST index in '%s'\n",
594                                m->opd_obd->obd_name, src);
595                         RETURN(-EINVAL);
596                 }
597                 m->opd_index = idx;
598                 m->opd_group = 0;
599                 idx = tgt - src;
600         } else {
601                 /* New OSC name fsname-OSTXXXX-osc-MDTXXXX */
602                 if (strncmp(tgt, "-MDT", 4) != 0 &&
603                          strncmp(tgt, "-OST", 4) != 0) {
604                         CERROR("%s: invalid target name %s\n",
605                                m->opd_obd->obd_name, lustre_cfg_string(cfg, 0));
606                         RETURN(-EINVAL);
607                 }
608
609                 idx = simple_strtol(tgt + 4, &mdt, 16);
610                 if (*mdt != '\0' || idx > INT_MAX || idx < 0) {
611                         CERROR("%s: invalid OST index in '%s'\n",
612                                m->opd_obd->obd_name, src);
613                         RETURN(-EINVAL);
614                 }
615
616                 /* Get MDT index from the name and set it to opd_group,
617                  * which will be used by OSP to connect with OST */
618                 m->opd_group = idx;
619                 if (tgt - src <= 12) {
620                         CERROR("%s: invalid mdt index retrieve from %s\n",
621                                m->opd_obd->obd_name, lustre_cfg_string(cfg, 0));
622                         RETURN(-EINVAL);
623                 }
624
625                 if (strncmp(tgt - 12, "-MDT", 4) == 0)
626                         m->opd_connect_mdt = 1;
627
628                 idx = simple_strtol(tgt - 8, &mdt, 16);
629                 if (mdt[0] != '-' || idx > INT_MAX || idx < 0) {
630                         CERROR("%s: invalid OST index in '%s'\n",
631                                m->opd_obd->obd_name, src);
632                         RETURN(-EINVAL);
633                 }
634
635                 m->opd_index = idx;
636                 idx = tgt - src - 12;
637         }
638         /* check the fsname length, and after this everything else will fit */
639         if (idx > MTI_NAME_MAXLEN) {
640                 CERROR("%s: fsname too long in '%s'\n",
641                        m->opd_obd->obd_name, src);
642                 RETURN(-EINVAL);
643         }
644
645         OBD_ALLOC(osdname, MAX_OBD_NAME);
646         if (osdname == NULL)
647                 RETURN(-ENOMEM);
648
649         memcpy(osdname, src, idx); /* copy just the fsname part */
650         osdname[idx] = '\0';
651
652         mdt = strstr(mdt, "-MDT");
653         if (mdt == NULL) /* 1.8 configs don't have "-MDT0000" at the end */
654                 strcat(osdname, "-MDT0000");
655         else
656                 strcat(osdname, mdt);
657         strcat(osdname, "-osd");
658         CDEBUG(D_HA, "%s: connect to %s (%s)\n", obd->obd_name, osdname, src);
659
660         if (m->opd_connect_mdt) {
661                 struct client_obd *cli = &m->opd_obd->u.cli;
662
663                 OBD_ALLOC(cli->cl_rpc_lock, sizeof(*cli->cl_rpc_lock));
664                 if (!cli->cl_rpc_lock)
665                         GOTO(out_fini, rc = -ENOMEM);
666                 osp_init_rpc_lock(cli->cl_rpc_lock);
667         }
668
669         m->opd_dt_dev.dd_lu_dev.ld_ops = &osp_lu_ops;
670         m->opd_dt_dev.dd_ops = &osp_dt_ops;
671         obd->obd_lu_dev = &m->opd_dt_dev.dd_lu_dev;
672
673         rc = osp_connect_to_osd(env, m, osdname);
674         if (rc)
675                 GOTO(out_fini, rc);
676
677         rc = ptlrpcd_addref();
678         if (rc)
679                 GOTO(out_disconnect, rc);
680
681         rc = client_obd_setup(obd, cfg);
682         if (rc) {
683                 CERROR("%s: can't setup obd: %d\n", m->opd_obd->obd_name, rc);
684                 GOTO(out_ref, rc);
685         }
686
687         osp_lprocfs_init(m);
688
689         if (!m->opd_connect_mdt) {
690                 /* Initialize last id from the storage - will be
691                  * used in orphan cleanup. */
692                 rc = osp_last_used_init(env, m);
693                 if (rc)
694                         GOTO(out_proc, rc);
695
696                 rc = obd_fid_init(m->opd_obd, NULL, LUSTRE_SEQ_DATA);
697                 if (rc) {
698                         CERROR("%s: fid init error: rc = %d\n",
699                                m->opd_obd->obd_name, rc);
700                         GOTO(out_last_used, rc);
701                 }
702
703                 /* Initialize precreation thread, it handles new
704                  * connections as well. */
705                 rc = osp_init_precreate(m);
706                 if (rc)
707                         GOTO(out_last_used, rc);
708                 /*
709                  * Initialize synhronization mechanism taking
710                  * care of propogating changes to OST in near
711                  * transactional manner.
712                  */
713                 rc = osp_sync_init(env, m);
714                 if (rc)
715                         GOTO(out_precreat, rc);
716
717         }
718         /*
719          * Initiate connect to OST
720          */
721         ll_generate_random_uuid(uuid);
722         class_uuid_unparse(uuid, &m->opd_cluuid);
723
724         imp = obd->u.cli.cl_import;
725
726         rc = ptlrpc_init_import(imp);
727         if (rc)
728                 GOTO(out, rc);
729         if (osdname)
730                 OBD_FREE(osdname, MAX_OBD_NAME);
731         RETURN(0);
732
733 out:
734         if (!m->opd_connect_mdt)
735                 /* stop sync thread */
736                 osp_sync_fini(m);
737 out_precreat:
738         /* stop precreate thread */
739         if (!m->opd_connect_mdt)
740                 osp_precreate_fini(m);
741 out_last_used:
742         if (!m->opd_connect_mdt)
743                 osp_last_used_fini(env, m);
744 out_proc:
745         ptlrpc_lprocfs_unregister_obd(obd);
746         lprocfs_obd_cleanup(obd);
747         obd_cleanup_client_import(obd);
748         if (m->opd_symlink)
749                 lprocfs_remove(&m->opd_symlink);
750         client_obd_cleanup(obd);
751 out_ref:
752         ptlrpcd_decref();
753 out_disconnect:
754         if (m->opd_connect_mdt) {
755                 struct client_obd *cli = &m->opd_obd->u.cli;
756                 if (cli->cl_rpc_lock != NULL) {
757                         OBD_FREE_PTR(cli->cl_rpc_lock);
758                         cli->cl_rpc_lock = NULL;
759                 }
760         }
761         obd_disconnect(m->opd_storage_exp);
762 out_fini:
763         if (osdname)
764                 OBD_FREE(osdname, MAX_OBD_NAME);
765         RETURN(rc);
766 }
767
768 static struct lu_device *osp_device_free(const struct lu_env *env,
769                                          struct lu_device *lu)
770 {
771         struct osp_device *m = lu2osp_dev(lu);
772
773         ENTRY;
774
775         if (cfs_atomic_read(&lu->ld_ref) && lu->ld_site) {
776                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
777                 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
778         }
779         dt_device_fini(&m->opd_dt_dev);
780         OBD_FREE_PTR(m);
781         RETURN(NULL);
782 }
783
784 static struct lu_device *osp_device_alloc(const struct lu_env *env,
785                                           struct lu_device_type *t,
786                                           struct lustre_cfg *lcfg)
787 {
788         struct osp_device *m;
789         struct lu_device  *l;
790
791         OBD_ALLOC_PTR(m);
792         if (m == NULL) {
793                 l = ERR_PTR(-ENOMEM);
794         } else {
795                 int rc;
796
797                 l = osp2lu_dev(m);
798                 dt_device_init(&m->opd_dt_dev, t);
799                 rc = osp_init0(env, m, t, lcfg);
800                 if (rc != 0) {
801                         osp_device_free(env, l);
802                         l = ERR_PTR(rc);
803                 }
804         }
805         return l;
806 }
807
808 static struct lu_device *osp_device_fini(const struct lu_env *env,
809                                          struct lu_device *d)
810 {
811         struct osp_device *m = lu2osp_dev(d);
812         struct obd_import *imp;
813         int                rc;
814
815         ENTRY;
816
817         if (m->opd_storage_exp)
818                 obd_disconnect(m->opd_storage_exp);
819
820         imp = m->opd_obd->u.cli.cl_import;
821
822         if (imp->imp_rq_pool) {
823                 ptlrpc_free_rq_pool(imp->imp_rq_pool);
824                 imp->imp_rq_pool = NULL;
825         }
826
827         obd_cleanup_client_import(m->opd_obd);
828
829         if (m->opd_symlink)
830                 lprocfs_remove(&m->opd_symlink);
831
832         LASSERT(m->opd_obd);
833         ptlrpc_lprocfs_unregister_obd(m->opd_obd);
834         lprocfs_obd_cleanup(m->opd_obd);
835
836         if (m->opd_connect_mdt) {
837                 struct client_obd *cli = &m->opd_obd->u.cli;
838                 if (cli->cl_rpc_lock != NULL) {
839                         OBD_FREE_PTR(cli->cl_rpc_lock);
840                         cli->cl_rpc_lock = NULL;
841                 }
842         }
843
844         rc = client_obd_cleanup(m->opd_obd);
845         LASSERTF(rc == 0, "error %d\n", rc);
846
847         ptlrpcd_decref();
848
849         RETURN(NULL);
850 }
851
852 static int osp_reconnect(const struct lu_env *env,
853                          struct obd_export *exp, struct obd_device *obd,
854                          struct obd_uuid *cluuid,
855                          struct obd_connect_data *data,
856                          void *localdata)
857 {
858         return 0;
859 }
860
861 /*
862  * we use exports to track all LOD users
863  */
864 static int osp_obd_connect(const struct lu_env *env, struct obd_export **exp,
865                            struct obd_device *obd, struct obd_uuid *cluuid,
866                            struct obd_connect_data *data, void *localdata)
867 {
868         struct osp_device       *osp = lu2osp_dev(obd->obd_lu_dev);
869         struct obd_connect_data *ocd;
870         struct obd_import       *imp;
871         struct lustre_handle     conn;
872         int                      rc;
873
874         ENTRY;
875
876         CDEBUG(D_CONFIG, "connect #%d\n", osp->opd_connects);
877
878         rc = class_connect(&conn, obd, cluuid);
879         if (rc)
880                 RETURN(rc);
881
882         *exp = class_conn2export(&conn);
883         /* Why should there ever be more than 1 connect? */
884         osp->opd_connects++;
885         LASSERT(osp->opd_connects == 1);
886
887         osp->opd_exp = *exp;
888
889         imp = osp->opd_obd->u.cli.cl_import;
890         imp->imp_dlm_handle = conn;
891
892         LASSERT(data != NULL);
893         LASSERT(data->ocd_connect_flags & OBD_CONNECT_INDEX);
894         ocd = &imp->imp_connect_data;
895         *ocd = *data;
896
897         imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
898
899         ocd->ocd_version = LUSTRE_VERSION_CODE;
900         ocd->ocd_index = data->ocd_index;
901         imp->imp_connect_flags_orig = ocd->ocd_connect_flags;
902
903         rc = ptlrpc_connect_import(imp);
904         if (rc) {
905                 CERROR("%s: can't connect obd: rc = %d\n", obd->obd_name, rc);
906                 GOTO(out, rc);
907         }
908
909         ptlrpc_pinger_add_import(imp);
910
911         if (osp->opd_connect_mdt && data->ocd_index == 0) {
912                 /* set seq controller export for MDC0 if exists */
913                 struct seq_server_site *ss;
914
915                 ss = lu_site2seq(osp2lu_dev(osp)->ld_site);
916                 ss->ss_control_exp = class_export_get(*exp);
917                 ss->ss_server_fld->lsf_control_exp = *exp;
918         }
919 out:
920         RETURN(rc);
921 }
922
923 /*
924  * once last export (we don't count self-export) disappeared
925  * osp can be released
926  */
927 static int osp_obd_disconnect(struct obd_export *exp)
928 {
929         struct obd_device *obd = exp->exp_obd;
930         struct osp_device *osp = lu2osp_dev(obd->obd_lu_dev);
931         int                rc;
932         ENTRY;
933
934         /* Only disconnect the underlying layers on the final disconnect. */
935         LASSERT(osp->opd_connects == 1);
936         osp->opd_connects--;
937
938         rc = class_disconnect(exp);
939         if (rc) {
940                 CERROR("%s: class disconnect error: rc = %d\n",
941                        obd->obd_name, rc);
942                 RETURN(rc);
943         }
944
945         /* destroy the device */
946         class_manual_cleanup(obd);
947
948         RETURN(rc);
949 }
950
951 /*
952  * lprocfs helpers still use OBD API, let's keep obd_statfs() support
953  */
954 static int osp_obd_statfs(const struct lu_env *env, struct obd_export *exp,
955                           struct obd_statfs *osfs, __u64 max_age, __u32 flags)
956 {
957         struct obd_statfs       *msfs;
958         struct ptlrpc_request   *req;
959         struct obd_import       *imp = NULL;
960         int                      rc;
961
962         ENTRY;
963
964         /* Since the request might also come from lprocfs, so we need
965          * sync this with client_disconnect_export Bug15684 */
966         down_read(&exp->exp_obd->u.cli.cl_sem);
967         if (exp->exp_obd->u.cli.cl_import)
968                 imp = class_import_get(exp->exp_obd->u.cli.cl_import);
969         up_read(&exp->exp_obd->u.cli.cl_sem);
970         if (!imp)
971                 RETURN(-ENODEV);
972
973         /* We could possibly pass max_age in the request (as an absolute
974          * timestamp or a "seconds.usec ago") so the target can avoid doing
975          * extra calls into the filesystem if that isn't necessary (e.g.
976          * during mount that would help a bit).  Having relative timestamps
977          * is not so great if request processing is slow, while absolute
978          * timestamps are not ideal because they need time synchronization. */
979         req = ptlrpc_request_alloc(imp, &RQF_OST_STATFS);
980
981         class_import_put(imp);
982
983         if (req == NULL)
984                 RETURN(-ENOMEM);
985
986         rc = ptlrpc_request_pack(req, LUSTRE_OST_VERSION, OST_STATFS);
987         if (rc) {
988                 ptlrpc_request_free(req);
989                 RETURN(rc);
990         }
991         ptlrpc_request_set_replen(req);
992         req->rq_request_portal = OST_CREATE_PORTAL;
993         ptlrpc_at_set_req_timeout(req);
994
995         if (flags & OBD_STATFS_NODELAY) {
996                 /* procfs requests not want stat in wait for avoid deadlock */
997                 req->rq_no_resend = 1;
998                 req->rq_no_delay = 1;
999         }
1000
1001         rc = ptlrpc_queue_wait(req);
1002         if (rc)
1003                 GOTO(out, rc);
1004
1005         msfs = req_capsule_server_get(&req->rq_pill, &RMF_OBD_STATFS);
1006         if (msfs == NULL)
1007                 GOTO(out, rc = -EPROTO);
1008
1009         *osfs = *msfs;
1010
1011         EXIT;
1012 out:
1013         ptlrpc_req_finished(req);
1014         return rc;
1015 }
1016
1017 static int osp_import_event(struct obd_device *obd, struct obd_import *imp,
1018                             enum obd_import_event event)
1019 {
1020         struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1021
1022         switch (event) {
1023         case IMP_EVENT_DISCON:
1024                 d->opd_got_disconnected = 1;
1025                 d->opd_imp_connected = 0;
1026                 if (d->opd_connect_mdt)
1027                         break;
1028                 osp_pre_update_status(d, -ENODEV);
1029                 cfs_waitq_signal(&d->opd_pre_waitq);
1030                 CDEBUG(D_HA, "got disconnected\n");
1031                 break;
1032         case IMP_EVENT_INACTIVE:
1033                 d->opd_imp_active = 0;
1034                 if (d->opd_connect_mdt)
1035                         break;
1036                 osp_pre_update_status(d, -ENODEV);
1037                 cfs_waitq_signal(&d->opd_pre_waitq);
1038                 CDEBUG(D_HA, "got inactive\n");
1039                 break;
1040         case IMP_EVENT_ACTIVE:
1041                 d->opd_imp_active = 1;
1042                 if (d->opd_got_disconnected)
1043                         d->opd_new_connection = 1;
1044                 d->opd_imp_connected = 1;
1045                 d->opd_imp_seen_connected = 1;
1046                 if (d->opd_connect_mdt)
1047                         break;
1048                 cfs_waitq_signal(&d->opd_pre_waitq);
1049                 __osp_sync_check_for_work(d);
1050                 CDEBUG(D_HA, "got connected\n");
1051                 break;
1052         case IMP_EVENT_INVALIDATE:
1053                 if (obd->obd_namespace == NULL)
1054                         break;
1055                 ldlm_namespace_cleanup(obd->obd_namespace, LDLM_FL_LOCAL_ONLY);
1056                 break;
1057         case IMP_EVENT_OCD:
1058         case IMP_EVENT_DEACTIVATE:
1059         case IMP_EVENT_ACTIVATE:
1060                 break;
1061         default:
1062                 CERROR("%s: unsupported import event: %#x\n",
1063                        obd->obd_name, event);
1064         }
1065         return 0;
1066 }
1067
1068 static int osp_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
1069                          void *karg, void *uarg)
1070 {
1071         struct obd_device       *obd = exp->exp_obd;
1072         struct osp_device       *d;
1073         struct obd_ioctl_data   *data = karg;
1074         int                      rc = 0;
1075
1076         ENTRY;
1077
1078         LASSERT(obd->obd_lu_dev);
1079         d = lu2osp_dev(obd->obd_lu_dev);
1080         LASSERT(d->opd_dt_dev.dd_ops == &osp_dt_ops);
1081
1082         if (!cfs_try_module_get(THIS_MODULE)) {
1083                 CERROR("%s: can't get module. Is it alive?", obd->obd_name);
1084                 return -EINVAL;
1085         }
1086
1087         switch (cmd) {
1088         case OBD_IOC_CLIENT_RECOVER:
1089                 rc = ptlrpc_recover_import(obd->u.cli.cl_import,
1090                                            data->ioc_inlbuf1, 0);
1091                 if (rc > 0)
1092                         rc = 0;
1093                 break;
1094         case IOC_OSC_SET_ACTIVE:
1095                 rc = ptlrpc_set_import_active(obd->u.cli.cl_import,
1096                                               data->ioc_offset);
1097                 break;
1098         case OBD_IOC_PING_TARGET:
1099                 rc = ptlrpc_obd_ping(obd);
1100                 break;
1101         default:
1102                 CERROR("%s: unrecognized ioctl %#x by %s\n", obd->obd_name,
1103                        cmd, cfs_curproc_comm());
1104                 rc = -ENOTTY;
1105         }
1106         cfs_module_put(THIS_MODULE);
1107         return rc;
1108 }
1109
1110 static int osp_obd_health_check(const struct lu_env *env,
1111                                 struct obd_device *obd)
1112 {
1113         struct osp_device *d = lu2osp_dev(obd->obd_lu_dev);
1114
1115         ENTRY;
1116
1117         /*
1118          * 1.8/2.0 behaviour is that OST being connected once at least
1119          * is considired "healthy". and one "healty" OST is enough to
1120          * allow lustre clients to connect to MDS
1121          */
1122         LASSERT(d);
1123         RETURN(!d->opd_imp_seen_connected);
1124 }
1125
1126 /* context key constructor/destructor: mdt_key_init, mdt_key_fini */
1127 LU_KEY_INIT_FINI(osp, struct osp_thread_info);
1128 static void osp_key_exit(const struct lu_context *ctx,
1129                          struct lu_context_key *key, void *data)
1130 {
1131         struct osp_thread_info *info = data;
1132
1133         info->osi_attr.la_valid = 0;
1134 }
1135
1136 struct lu_context_key osp_thread_key = {
1137         .lct_tags = LCT_MD_THREAD,
1138         .lct_init = osp_key_init,
1139         .lct_fini = osp_key_fini,
1140         .lct_exit = osp_key_exit
1141 };
1142
1143 /* context key constructor/destructor: mdt_txn_key_init, mdt_txn_key_fini */
1144 LU_KEY_INIT_FINI(osp_txn, struct osp_txn_info);
1145
1146 struct lu_context_key osp_txn_key = {
1147         .lct_tags = LCT_OSP_THREAD,
1148         .lct_init = osp_txn_key_init,
1149         .lct_fini = osp_txn_key_fini
1150 };
1151 LU_TYPE_INIT_FINI(osp, &osp_thread_key, &osp_txn_key);
1152
1153 static struct lu_device_type_operations osp_device_type_ops = {
1154         .ldto_init           = osp_type_init,
1155         .ldto_fini           = osp_type_fini,
1156
1157         .ldto_start          = osp_type_start,
1158         .ldto_stop           = osp_type_stop,
1159
1160         .ldto_device_alloc   = osp_device_alloc,
1161         .ldto_device_free    = osp_device_free,
1162
1163         .ldto_device_fini    = osp_device_fini
1164 };
1165
1166 static struct lu_device_type osp_device_type = {
1167         .ldt_tags     = LU_DEVICE_DT,
1168         .ldt_name     = LUSTRE_OSP_NAME,
1169         .ldt_ops      = &osp_device_type_ops,
1170         .ldt_ctx_tags = LCT_MD_THREAD
1171 };
1172
1173 static struct obd_ops osp_obd_device_ops = {
1174         .o_owner        = THIS_MODULE,
1175         .o_add_conn     = client_import_add_conn,
1176         .o_del_conn     = client_import_del_conn,
1177         .o_reconnect    = osp_reconnect,
1178         .o_connect      = osp_obd_connect,
1179         .o_disconnect   = osp_obd_disconnect,
1180         .o_health_check = osp_obd_health_check,
1181         .o_import_event = osp_import_event,
1182         .o_iocontrol    = osp_iocontrol,
1183         .o_statfs       = osp_obd_statfs,
1184         .o_fid_init     = client_fid_init,
1185         .o_fid_fini     = client_fid_fini,
1186 };
1187
1188 struct llog_operations osp_mds_ost_orig_logops;
1189
1190 static int __init osp_mod_init(void)
1191 {
1192         struct lprocfs_static_vars       lvars;
1193         cfs_proc_dir_entry_t            *osc_proc_dir;
1194         int                              rc;
1195
1196         rc = lu_kmem_init(osp_caches);
1197         if (rc)
1198                 return rc;
1199
1200         lprocfs_osp_init_vars(&lvars);
1201
1202         rc = class_register_type(&osp_obd_device_ops, NULL, lvars.module_vars,
1203                                  LUSTRE_OSP_NAME, &osp_device_type);
1204
1205         /* create "osc" entry in procfs for compatibility purposes */
1206         if (rc != 0) {
1207                 lu_kmem_fini(osp_caches);
1208                 return rc;
1209         }
1210
1211         lprocfs_lwp_init_vars(&lvars);
1212
1213         rc = class_register_type(&lwp_obd_device_ops, NULL, lvars.module_vars,
1214                                  LUSTRE_LWP_NAME, &lwp_device_type);
1215         if (rc != 0) {
1216                 class_unregister_type(LUSTRE_OSP_NAME);
1217                 lu_kmem_fini(osp_caches);
1218                 return rc;
1219         }
1220
1221         /* Note: add_rec/delcare_add_rec will be only used by catalogs */
1222         osp_mds_ost_orig_logops = llog_osd_ops;
1223         osp_mds_ost_orig_logops.lop_add = llog_cat_add_rec;
1224         osp_mds_ost_orig_logops.lop_declare_add = llog_cat_declare_add_rec;
1225
1226         osc_proc_dir = lprocfs_srch(proc_lustre_root, "osc");
1227         if (osc_proc_dir == NULL) {
1228                 osc_proc_dir = lprocfs_register("osc", proc_lustre_root, NULL,
1229                                                 NULL);
1230                 if (IS_ERR(osc_proc_dir))
1231                         CERROR("osp: can't create compat entry \"osc\": %d\n",
1232                                (int) PTR_ERR(osc_proc_dir));
1233         }
1234         return rc;
1235 }
1236
1237 static void __exit osp_mod_exit(void)
1238 {
1239         lprocfs_try_remove_proc_entry("osc", proc_lustre_root);
1240
1241         class_unregister_type(LUSTRE_LWP_NAME);
1242         class_unregister_type(LUSTRE_OSP_NAME);
1243         lu_kmem_fini(osp_caches);
1244 }
1245
1246 MODULE_AUTHOR("Intel, Inc. <http://www.intel.com/>");
1247 MODULE_DESCRIPTION("Lustre OST Proxy Device ("LUSTRE_OSP_NAME")");
1248 MODULE_LICENSE("GPL");
1249
1250 cfs_module(osp, LUSTRE_VERSION_STRING, osp_mod_init, osp_mod_exit);