Whamcloud - gitweb
LU-911 osd: use struct obd_statfs instead of kstatfs_t
[fs/lustre-release.git] / lustre / cmm / cmm_device.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * GPL HEADER START
5  *
6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
7  *
8  * This program is free software; you can redistribute it and/or modify
9  * it under the terms of the GNU General Public License version 2 only,
10  * as published by the Free Software Foundation.
11  *
12  * This program is distributed in the hope that it will be useful, but
13  * WITHOUT ANY WARRANTY; without even the implied warranty of
14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
15  * General Public License version 2 for more details (a copy is included
16  * in the LICENSE file that accompanied this code).
17  *
18  * You should have received a copy of the GNU General Public License
19  * version 2 along with this program; If not, see
20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
21  *
22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23  * CA 95054 USA or visit www.sun.com if you need additional information or
24  * have any questions.
25  *
26  * GPL HEADER END
27  */
28 /*
29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
30  * Use is subject to license terms.
31  *
32  * Copyright (c) 2011, Whamcloud, Inc.
33  */
34 /*
35  * This file is part of Lustre, http://www.lustre.org/
36  * Lustre is a trademark of Sun Microsystems, Inc.
37  *
38  * lustre/cmm/cmm_device.c
39  *
40  * Lustre Cluster Metadata Manager (cmm)
41  *
42  * Author: Mike Pershin <tappro@clusterfs.com>
43  */
44 /**
45  * \addtogroup cmm
46  * @{
47  */
48 #ifndef EXPORT_SYMTAB
49 # define EXPORT_SYMTAB
50 #endif
51 #define DEBUG_SUBSYSTEM S_MDS
52
53 #include <linux/module.h>
54
55 #include <obd.h>
56 #include <obd_class.h>
57 #include <lprocfs_status.h>
58 #include <lustre_ver.h>
59 #include "cmm_internal.h"
60 #include "mdc_internal.h"
61 #ifdef HAVE_QUOTA_SUPPORT
62 # include <lustre_quota.h>
63 #endif
64
65 struct obd_ops cmm_obd_device_ops = {
66         .o_owner           = THIS_MODULE
67 };
68
69 static const struct lu_device_operations cmm_lu_ops;
70
71 static inline int lu_device_is_cmm(struct lu_device *d)
72 {
73         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &cmm_lu_ops);
74 }
75
76 int cmm_root_get(const struct lu_env *env, struct md_device *md,
77                  struct lu_fid *fid)
78 {
79         struct cmm_device *cmm_dev = md2cmm_dev(md);
80         /* valid only on master MDS */
81         if (cmm_dev->cmm_local_num == 0)
82                 return cmm_child_ops(cmm_dev)->mdo_root_get(env,
83                                      cmm_dev->cmm_child, fid);
84         else
85                 return -EINVAL;
86 }
87
88 static int cmm_statfs(const struct lu_env *env, struct md_device *md,
89                       struct obd_statfs *sfs)
90 {
91         struct cmm_device *cmm_dev = md2cmm_dev(md);
92         int rc;
93
94         ENTRY;
95         rc = cmm_child_ops(cmm_dev)->mdo_statfs(env,
96                                                 cmm_dev->cmm_child, sfs);
97         RETURN (rc);
98 }
99
100 static int cmm_maxsize_get(const struct lu_env *env, struct md_device *md,
101                            int *md_size, int *cookie_size)
102 {
103         struct cmm_device *cmm_dev = md2cmm_dev(md);
104         int rc;
105         ENTRY;
106         rc = cmm_child_ops(cmm_dev)->mdo_maxsize_get(env, cmm_dev->cmm_child,
107                                                      md_size, cookie_size);
108         RETURN(rc);
109 }
110
111 static int cmm_init_capa_ctxt(const struct lu_env *env, struct md_device *md,
112                               int mode , unsigned long timeout, __u32 alg,
113                               struct lustre_capa_key *keys)
114 {
115         struct cmm_device *cmm_dev = md2cmm_dev(md);
116         int rc;
117         ENTRY;
118         LASSERT(cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt);
119         rc = cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt(env, cmm_dev->cmm_child,
120                                                         mode, timeout, alg,
121                                                         keys);
122         RETURN(rc);
123 }
124
125 static int cmm_update_capa_key(const struct lu_env *env,
126                                struct md_device *md,
127                                struct lustre_capa_key *key)
128 {
129         struct cmm_device *cmm_dev = md2cmm_dev(md);
130         int rc;
131         ENTRY;
132         rc = cmm_child_ops(cmm_dev)->mdo_update_capa_key(env,
133                                                          cmm_dev->cmm_child,
134                                                          key);
135         RETURN(rc);
136 }
137
138 static int cmm_llog_ctxt_get(const struct lu_env *env, struct md_device *m,
139                              int idx, void **h)
140 {
141         struct cmm_device *cmm_dev = md2cmm_dev(m);
142         int rc;
143         ENTRY;
144
145         rc = cmm_child_ops(cmm_dev)->mdo_llog_ctxt_get(env, cmm_dev->cmm_child,
146                                                        idx, h);
147         RETURN(rc);
148 }
149
150 #ifdef HAVE_QUOTA_SUPPORT
151 /**
152  * \name Quota functions
153  * @{
154  */
155 static int cmm_quota_notify(const struct lu_env *env, struct md_device *m)
156 {
157         struct cmm_device *cmm_dev = md2cmm_dev(m);
158         int rc;
159         ENTRY;
160
161         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_notify(env,
162                                                           cmm_dev->cmm_child);
163         RETURN(rc);
164 }
165
166 static int cmm_quota_setup(const struct lu_env *env, struct md_device *m,
167                            void *data)
168 {
169         struct cmm_device *cmm_dev = md2cmm_dev(m);
170         int rc;
171         ENTRY;
172
173         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_setup(env,
174                                                          cmm_dev->cmm_child,
175                                                          data);
176         RETURN(rc);
177 }
178
179 static int cmm_quota_cleanup(const struct lu_env *env, struct md_device *m)
180 {
181         struct cmm_device *cmm_dev = md2cmm_dev(m);
182         int rc;
183         ENTRY;
184
185         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_cleanup(env,
186                                                            cmm_dev->cmm_child);
187         RETURN(rc);
188 }
189
190 static int cmm_quota_recovery(const struct lu_env *env, struct md_device *m)
191 {
192         struct cmm_device *cmm_dev = md2cmm_dev(m);
193         int rc;
194         ENTRY;
195
196         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_recovery(env,
197                                                             cmm_dev->cmm_child);
198         RETURN(rc);
199 }
200
201 static int cmm_quota_check(const struct lu_env *env, struct md_device *m,
202                            __u32 type)
203 {
204         struct cmm_device *cmm_dev = md2cmm_dev(m);
205         int rc;
206         ENTRY;
207
208         /* disable quota for CMD case temporary. */
209         if (cmm_dev->cmm_tgt_count)
210                 RETURN(-EOPNOTSUPP);
211
212         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_check(env,
213                                                          cmm_dev->cmm_child,
214                                                          type);
215         RETURN(rc);
216 }
217
218 static int cmm_quota_on(const struct lu_env *env, struct md_device *m,
219                         __u32 type)
220 {
221         struct cmm_device *cmm_dev = md2cmm_dev(m);
222         int rc;
223         ENTRY;
224
225         /* disable quota for CMD case temporary. */
226         if (cmm_dev->cmm_tgt_count)
227                 RETURN(-EOPNOTSUPP);
228
229         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_on(env,
230                                                       cmm_dev->cmm_child,
231                                                       type);
232         RETURN(rc);
233 }
234
235 static int cmm_quota_off(const struct lu_env *env, struct md_device *m,
236                          __u32 type)
237 {
238         struct cmm_device *cmm_dev = md2cmm_dev(m);
239         int rc;
240         ENTRY;
241
242         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_off(env,
243                                                        cmm_dev->cmm_child,
244                                                        type);
245         RETURN(rc);
246 }
247
248 static int cmm_quota_setinfo(const struct lu_env *env, struct md_device *m,
249                              __u32 type, __u32 id, struct obd_dqinfo *dqinfo)
250 {
251         struct cmm_device *cmm_dev = md2cmm_dev(m);
252         int rc;
253         ENTRY;
254
255         /* disable quota for CMD case temporary. */
256         if (cmm_dev->cmm_tgt_count)
257                 RETURN(-EOPNOTSUPP);
258
259         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_setinfo(env,
260                                                            cmm_dev->cmm_child,
261                                                            type, id, dqinfo);
262         RETURN(rc);
263 }
264
265 static int cmm_quota_getinfo(const struct lu_env *env,
266                              const struct md_device *m,
267                              __u32 type, __u32 id, struct obd_dqinfo *dqinfo)
268 {
269         struct cmm_device *cmm_dev = md2cmm_dev((struct md_device *)m);
270         int rc;
271         ENTRY;
272
273         /* disable quota for CMD case temporary. */
274         if (cmm_dev->cmm_tgt_count)
275                 RETURN(-EOPNOTSUPP);
276
277         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_getinfo(env,
278                                                            cmm_dev->cmm_child,
279                                                            type, id, dqinfo);
280         RETURN(rc);
281 }
282
283 static int cmm_quota_setquota(const struct lu_env *env, struct md_device *m,
284                               __u32 type, __u32 id, struct obd_dqblk *dqblk)
285 {
286         struct cmm_device *cmm_dev = md2cmm_dev(m);
287         int rc;
288         ENTRY;
289
290         /* disable quota for CMD case temporary. */
291         if (cmm_dev->cmm_tgt_count)
292                 RETURN(-EOPNOTSUPP);
293
294         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_setquota(env,
295                                                             cmm_dev->cmm_child,
296                                                             type, id, dqblk);
297         RETURN(rc);
298 }
299
300 static int cmm_quota_getquota(const struct lu_env *env,
301                               const struct md_device *m,
302                               __u32 type, __u32 id, struct obd_dqblk *dqblk)
303 {
304         struct cmm_device *cmm_dev = md2cmm_dev((struct md_device *)m);
305         int rc;
306         ENTRY;
307
308         /* disable quota for CMD case temporary. */
309         if (cmm_dev->cmm_tgt_count)
310                 RETURN(-EOPNOTSUPP);
311
312         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_getquota(env,
313                                                             cmm_dev->cmm_child,
314                                                             type, id, dqblk);
315         RETURN(rc);
316 }
317
318 static int cmm_quota_getoinfo(const struct lu_env *env,
319                               const struct md_device *m,
320                               __u32 type, __u32 id, struct obd_dqinfo *dqinfo)
321 {
322         struct cmm_device *cmm_dev = md2cmm_dev((struct md_device *)m);
323         int rc;
324         ENTRY;
325
326         /* disable quota for CMD case temporary. */
327         if (cmm_dev->cmm_tgt_count)
328                 RETURN(-EOPNOTSUPP);
329
330         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_getoinfo(env,
331                                                             cmm_dev->cmm_child,
332                                                             type, id, dqinfo);
333         RETURN(rc);
334 }
335
336 static int cmm_quota_getoquota(const struct lu_env *env,
337                                const struct md_device *m,
338                                __u32 type, __u32 id, struct obd_dqblk *dqblk)
339 {
340         struct cmm_device *cmm_dev = md2cmm_dev((struct md_device *)m);
341         int rc;
342         ENTRY;
343
344         /* disable quota for CMD case temporary. */
345         if (cmm_dev->cmm_tgt_count)
346                 RETURN(-EOPNOTSUPP);
347
348         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_getoquota(env,
349                                                              cmm_dev->cmm_child,
350                                                              type, id, dqblk);
351         RETURN(rc);
352 }
353
354 static int cmm_quota_invalidate(const struct lu_env *env, struct md_device *m,
355                                 __u32 type)
356 {
357         struct cmm_device *cmm_dev = md2cmm_dev(m);
358         int rc;
359         ENTRY;
360
361         /* disable quota for CMD case temporary. */
362         if (cmm_dev->cmm_tgt_count)
363                 RETURN(-EOPNOTSUPP);
364
365         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_invalidate(env,
366                                                               cmm_dev->cmm_child,
367                                                               type);
368         RETURN(rc);
369 }
370
371 static int cmm_quota_finvalidate(const struct lu_env *env, struct md_device *m,
372                                  __u32 type)
373 {
374         struct cmm_device *cmm_dev = md2cmm_dev(m);
375         int rc;
376         ENTRY;
377
378         /* disable quota for CMD case temporary. */
379         if (cmm_dev->cmm_tgt_count)
380                 RETURN(-EOPNOTSUPP);
381
382         rc = cmm_child_ops(cmm_dev)->mdo_quota.mqo_finvalidate(env,
383                                                                cmm_dev->cmm_child,
384                                                                type);
385         RETURN(rc);
386 }
387 /** @} */
388 #endif
389
390 int cmm_iocontrol(const struct lu_env *env, struct md_device *m,
391                   unsigned int cmd, int len, void *data)
392 {
393         struct md_device *next = md2cmm_dev(m)->cmm_child;
394         int rc;
395
396         ENTRY;
397         rc = next->md_ops->mdo_iocontrol(env, next, cmd, len, data);
398         RETURN(rc);
399 }
400
401
402 static const struct md_device_operations cmm_md_ops = {
403         .mdo_statfs          = cmm_statfs,
404         .mdo_root_get        = cmm_root_get,
405         .mdo_maxsize_get     = cmm_maxsize_get,
406         .mdo_init_capa_ctxt  = cmm_init_capa_ctxt,
407         .mdo_update_capa_key = cmm_update_capa_key,
408         .mdo_llog_ctxt_get   = cmm_llog_ctxt_get,
409         .mdo_iocontrol       = cmm_iocontrol,
410 #ifdef HAVE_QUOTA_SUPPORT
411         .mdo_quota           = {
412                 .mqo_notify      = cmm_quota_notify,
413                 .mqo_setup       = cmm_quota_setup,
414                 .mqo_cleanup     = cmm_quota_cleanup,
415                 .mqo_recovery    = cmm_quota_recovery,
416                 .mqo_check       = cmm_quota_check,
417                 .mqo_on          = cmm_quota_on,
418                 .mqo_off         = cmm_quota_off,
419                 .mqo_setinfo     = cmm_quota_setinfo,
420                 .mqo_getinfo     = cmm_quota_getinfo,
421                 .mqo_setquota    = cmm_quota_setquota,
422                 .mqo_getquota    = cmm_quota_getquota,
423                 .mqo_getoinfo    = cmm_quota_getoinfo,
424                 .mqo_getoquota   = cmm_quota_getoquota,
425                 .mqo_invalidate  = cmm_quota_invalidate,
426                 .mqo_finvalidate = cmm_quota_finvalidate
427         }
428 #endif
429 };
430
431 extern struct lu_device_type mdc_device_type;
432 /**
433  * Init MDC.
434  */
435 static int cmm_post_init_mdc(const struct lu_env *env,
436                              struct cmm_device *cmm)
437 {
438         int max_mdsize, max_cookiesize, rc;
439         struct mdc_device *mc, *tmp;
440
441         /* get the max mdsize and cookiesize from lower layer */
442         rc = cmm_maxsize_get(env, &cmm->cmm_md_dev, &max_mdsize,
443                              &max_cookiesize);
444         if (rc)
445                 RETURN(rc);
446
447         cfs_spin_lock(&cmm->cmm_tgt_guard);
448         cfs_list_for_each_entry_safe(mc, tmp, &cmm->cmm_targets,
449                                      mc_linkage) {
450                 cmm_mdc_init_ea_size(env, mc, max_mdsize, max_cookiesize);
451         }
452         cfs_spin_unlock(&cmm->cmm_tgt_guard);
453         RETURN(rc);
454 }
455
456 /* --- cmm_lu_operations --- */
457 /* add new MDC to the CMM, create MDC lu_device and connect it to mdc_obd */
458 static int cmm_add_mdc(const struct lu_env *env,
459                        struct cmm_device *cm, struct lustre_cfg *cfg)
460 {
461         struct lu_device_type *ldt = &mdc_device_type;
462         char *p, *num = lustre_cfg_string(cfg, 2);
463         struct mdc_device *mc, *tmp;
464         struct lu_fld_target target;
465         struct lu_device *ld;
466         struct lu_device *cmm_lu = cmm2lu_dev(cm);
467         mdsno_t mdc_num;
468         struct lu_site *site = cmm2lu_dev(cm)->ld_site;
469         int rc;
470 #ifdef HAVE_QUOTA_SUPPORT
471         int first;
472 #endif
473         ENTRY;
474
475         /* find out that there is no such mdc */
476         LASSERT(num);
477         mdc_num = simple_strtol(num, &p, 10);
478         if (*p) {
479                 CERROR("Invalid index in lustre_cgf, offset 2\n");
480                 RETURN(-EINVAL);
481         }
482
483         cfs_spin_lock(&cm->cmm_tgt_guard);
484         cfs_list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
485                                      mc_linkage) {
486                 if (mc->mc_num == mdc_num) {
487                         cfs_spin_unlock(&cm->cmm_tgt_guard);
488                         RETURN(-EEXIST);
489                 }
490         }
491         cfs_spin_unlock(&cm->cmm_tgt_guard);
492         ld = ldt->ldt_ops->ldto_device_alloc(env, ldt, cfg);
493         if (IS_ERR(ld))
494                 RETURN(PTR_ERR(ld));
495
496         ld->ld_site = site;
497
498         rc = ldt->ldt_ops->ldto_device_init(env, ld, NULL, NULL);
499         if (rc) {
500                 ldt->ldt_ops->ldto_device_free(env, ld);
501                 RETURN(rc);
502         }
503         /* pass config to the just created MDC */
504         rc = ld->ld_ops->ldo_process_config(env, ld, cfg);
505         if (rc) {
506                 ldt->ldt_ops->ldto_device_fini(env, ld);
507                 ldt->ldt_ops->ldto_device_free(env, ld);
508                 RETURN(rc);
509         }
510
511         cfs_spin_lock(&cm->cmm_tgt_guard);
512         cfs_list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
513                                      mc_linkage) {
514                 if (mc->mc_num == mdc_num) {
515                         cfs_spin_unlock(&cm->cmm_tgt_guard);
516                         ldt->ldt_ops->ldto_device_fini(env, ld);
517                         ldt->ldt_ops->ldto_device_free(env, ld);
518                         RETURN(-EEXIST);
519                 }
520         }
521         mc = lu2mdc_dev(ld);
522         cfs_list_add_tail(&mc->mc_linkage, &cm->cmm_targets);
523         cm->cmm_tgt_count++;
524 #ifdef HAVE_QUOTA_SUPPORT
525         first = cm->cmm_tgt_count;
526 #endif
527         cfs_spin_unlock(&cm->cmm_tgt_guard);
528
529         lu_device_get(cmm_lu);
530         lu_ref_add(&cmm_lu->ld_reference, "mdc-child", ld);
531
532         target.ft_srv = NULL;
533         target.ft_idx = mc->mc_num;
534         target.ft_exp = mc->mc_desc.cl_exp;
535         fld_client_add_target(cm->cmm_fld, &target);
536
537         if (mc->mc_num == 0) {
538                 /* this is mdt0 -> mc export, fld lookup need this export
539                    to forward fld lookup request. */
540                 LASSERT(!lu_site2md(site)->ms_server_fld->lsf_control_exp);
541                 lu_site2md(site)->ms_server_fld->lsf_control_exp =
542                                           mc->mc_desc.cl_exp;
543         }
544 #ifdef HAVE_QUOTA_SUPPORT
545         /* XXX: Disable quota for CMD case temporary. */
546         if (first == 1) {
547                 CWARN("Disable quota for CMD case temporary!\n");
548                 cmm_child_ops(cm)->mdo_quota.mqo_off(env, cm->cmm_child, UGQUOTA);
549         }
550 #endif
551         /* Set max md size for the mdc. */
552         rc = cmm_post_init_mdc(env, cm);
553         RETURN(rc);
554 }
555
556 static void cmm_device_shutdown(const struct lu_env *env,
557                                 struct cmm_device *cm,
558                                 struct lustre_cfg *cfg)
559 {
560         struct mdc_device *mc, *tmp;
561         ENTRY;
562
563         /* Remove local target from FLD. */
564         fld_client_del_target(cm->cmm_fld, cm->cmm_local_num);
565
566         /* Finish all mdc devices. */
567         cfs_spin_lock(&cm->cmm_tgt_guard);
568         cfs_list_for_each_entry_safe(mc, tmp, &cm->cmm_targets, mc_linkage) {
569                 struct lu_device *ld_m = mdc2lu_dev(mc);
570                 fld_client_del_target(cm->cmm_fld, mc->mc_num);
571                 ld_m->ld_ops->ldo_process_config(env, ld_m, cfg);
572         }
573         cfs_spin_unlock(&cm->cmm_tgt_guard);
574
575         /* remove upcall device*/
576         md_upcall_fini(&cm->cmm_md_dev);
577
578         EXIT;
579 }
580
581 static int cmm_device_mount(const struct lu_env *env,
582                             struct cmm_device *m, struct lustre_cfg *cfg)
583 {
584         const char *index = lustre_cfg_string(cfg, 2);
585         char *p;
586
587         LASSERT(index != NULL);
588
589         m->cmm_local_num = simple_strtol(index, &p, 10);
590         if (*p) {
591                 CERROR("Invalid index in lustre_cgf\n");
592                 RETURN(-EINVAL);
593         }
594
595         RETURN(0);
596 }
597
598 static int cmm_process_config(const struct lu_env *env,
599                               struct lu_device *d, struct lustre_cfg *cfg)
600 {
601         struct cmm_device *m = lu2cmm_dev(d);
602         struct lu_device *next = md2lu_dev(m->cmm_child);
603         int err;
604         ENTRY;
605
606         switch(cfg->lcfg_command) {
607         case LCFG_ADD_MDC:
608                 /* On first ADD_MDC add also local target. */
609                 if (!(m->cmm_flags & CMM_INITIALIZED)) {
610                         struct lu_site *ls = cmm2lu_dev(m)->ld_site;
611                         struct lu_fld_target target;
612
613                         target.ft_srv = lu_site2md(ls)->ms_server_fld;
614                         target.ft_idx = m->cmm_local_num;
615                         target.ft_exp = NULL;
616
617                         fld_client_add_target(m->cmm_fld, &target);
618                 }
619                 err = cmm_add_mdc(env, m, cfg);
620
621                 /* The first ADD_MDC can be counted as setup is finished. */
622                 if (!(m->cmm_flags & CMM_INITIALIZED))
623                         m->cmm_flags |= CMM_INITIALIZED;
624
625                 break;
626         case LCFG_SETUP:
627         {
628                 /* lower layers should be set up at first */
629                 err = next->ld_ops->ldo_process_config(env, next, cfg);
630                 if (err == 0)
631                         err = cmm_device_mount(env, m, cfg);
632                 break;
633         }
634         case LCFG_CLEANUP:
635         {
636                 cmm_device_shutdown(env, m, cfg);
637         }
638         default:
639                 err = next->ld_ops->ldo_process_config(env, next, cfg);
640         }
641         RETURN(err);
642 }
643
644 static int cmm_recovery_complete(const struct lu_env *env,
645                                  struct lu_device *d)
646 {
647         struct cmm_device *m = lu2cmm_dev(d);
648         struct lu_device *next = md2lu_dev(m->cmm_child);
649         int rc;
650         ENTRY;
651         rc = next->ld_ops->ldo_recovery_complete(env, next);
652         RETURN(rc);
653 }
654
655 static int cmm_prepare(const struct lu_env *env,
656                        struct lu_device *pdev,
657                        struct lu_device *dev)
658 {
659         struct cmm_device *cmm = lu2cmm_dev(dev);
660         struct lu_device *next = md2lu_dev(cmm->cmm_child);
661         int rc;
662
663         ENTRY;
664         rc = next->ld_ops->ldo_prepare(env, dev, next);
665         RETURN(rc);
666 }
667
668 static const struct lu_device_operations cmm_lu_ops = {
669         .ldo_object_alloc      = cmm_object_alloc,
670         .ldo_process_config    = cmm_process_config,
671         .ldo_recovery_complete = cmm_recovery_complete,
672         .ldo_prepare           = cmm_prepare,
673 };
674
675 /* --- lu_device_type operations --- */
676 int cmm_upcall(const struct lu_env *env, struct md_device *md,
677                enum md_upcall_event ev, void *data)
678 {
679         int rc;
680         ENTRY;
681
682         switch (ev) {
683                 case MD_LOV_SYNC:
684                         rc = cmm_post_init_mdc(env, md2cmm_dev(md));
685                         if (rc)
686                                 CERROR("can not init md size %d\n", rc);
687                         /* fall through */
688                 default:
689                         rc = md_do_upcall(env, md, ev, data);
690         }
691         RETURN(rc);
692 }
693
694 static struct lu_device *cmm_device_free(const struct lu_env *env,
695                                          struct lu_device *d)
696 {
697         struct cmm_device *m = lu2cmm_dev(d);
698         struct lu_device  *next = md2lu_dev(m->cmm_child);
699         ENTRY;
700
701         LASSERT(m->cmm_tgt_count == 0);
702         LASSERT(cfs_list_empty(&m->cmm_targets));
703         if (m->cmm_fld != NULL) {
704                 OBD_FREE_PTR(m->cmm_fld);
705                 m->cmm_fld = NULL;
706         }
707         md_device_fini(&m->cmm_md_dev);
708         OBD_FREE_PTR(m);
709         RETURN(next);
710 }
711
712 static struct lu_device *cmm_device_alloc(const struct lu_env *env,
713                                           struct lu_device_type *t,
714                                           struct lustre_cfg *cfg)
715 {
716         struct lu_device  *l;
717         struct cmm_device *m;
718         ENTRY;
719
720         OBD_ALLOC_PTR(m);
721         if (m == NULL) {
722                 l = ERR_PTR(-ENOMEM);
723         } else {
724                 md_device_init(&m->cmm_md_dev, t);
725                 m->cmm_md_dev.md_ops = &cmm_md_ops;
726                 md_upcall_init(&m->cmm_md_dev, cmm_upcall);
727                 l = cmm2lu_dev(m);
728                 l->ld_ops = &cmm_lu_ops;
729
730                 OBD_ALLOC_PTR(m->cmm_fld);
731                 if (!m->cmm_fld) {
732                         cmm_device_free(env, l);
733                         l = ERR_PTR(-ENOMEM);
734                 }
735         }
736         RETURN(l);
737 }
738
739 /* context key constructor/destructor: cmm_key_init, cmm_key_fini */
740 LU_KEY_INIT_FINI(cmm, struct cmm_thread_info);
741
742 /* context key: cmm_thread_key */
743 LU_CONTEXT_KEY_DEFINE(cmm, LCT_MD_THREAD);
744
745 struct cmm_thread_info *cmm_env_info(const struct lu_env *env)
746 {
747         struct cmm_thread_info *info;
748
749         info = lu_context_key_get(&env->le_ctx, &cmm_thread_key);
750         LASSERT(info != NULL);
751         return info;
752 }
753
754 /* type constructor/destructor: cmm_type_init/cmm_type_fini */
755 LU_TYPE_INIT_FINI(cmm, &cmm_thread_key);
756
757 /* 
758  * Kludge code : it should be moved mdc_device.c if mdc_(mds)_device
759  * is really stacked.
760  */
761 static int __cmm_type_init(struct lu_device_type *t)
762 {
763         int rc;
764         rc = lu_device_type_init(&mdc_device_type);
765         if (rc == 0) {
766                 rc = cmm_type_init(t);
767                 if (rc)
768                         lu_device_type_fini(&mdc_device_type);
769         }
770         return rc;
771 }
772
773 static void __cmm_type_fini(struct lu_device_type *t)
774 {
775         lu_device_type_fini(&mdc_device_type);
776         cmm_type_fini(t);
777 }
778
779 static void __cmm_type_start(struct lu_device_type *t)
780 {
781         mdc_device_type.ldt_ops->ldto_start(&mdc_device_type);
782         cmm_type_start(t);
783 }
784
785 static void __cmm_type_stop(struct lu_device_type *t)
786 {
787         mdc_device_type.ldt_ops->ldto_stop(&mdc_device_type);
788         cmm_type_stop(t);
789 }
790
791 static int cmm_device_init(const struct lu_env *env, struct lu_device *d,
792                            const char *name, struct lu_device *next)
793 {
794         struct cmm_device *m = lu2cmm_dev(d);
795         struct lu_site *ls;
796         int err = 0;
797         ENTRY;
798
799         cfs_spin_lock_init(&m->cmm_tgt_guard);
800         CFS_INIT_LIST_HEAD(&m->cmm_targets);
801         m->cmm_tgt_count = 0;
802         m->cmm_child = lu2md_dev(next);
803
804         err = fld_client_init(m->cmm_fld, name,
805                               LUSTRE_CLI_FLD_HASH_DHT);
806         if (err) {
807                 CERROR("Can't init FLD, err %d\n", err);
808                 RETURN(err);
809         }
810
811         /* Assign site's fld client ref, needed for asserts in osd. */
812         ls = cmm2lu_dev(m)->ld_site;
813         lu_site2md(ls)->ms_client_fld = m->cmm_fld;
814         err = cmm_procfs_init(m, name);
815
816         RETURN(err);
817 }
818
819 static struct lu_device *cmm_device_fini(const struct lu_env *env,
820                                          struct lu_device *ld)
821 {
822         struct cmm_device *cm = lu2cmm_dev(ld);
823         struct mdc_device *mc, *tmp;
824         struct lu_site *ls;
825         ENTRY;
826
827         /* Finish all mdc devices */
828         cfs_spin_lock(&cm->cmm_tgt_guard);
829         cfs_list_for_each_entry_safe(mc, tmp, &cm->cmm_targets, mc_linkage) {
830                 struct lu_device *ld_m = mdc2lu_dev(mc);
831                 struct lu_device *ld_c = cmm2lu_dev(cm);
832
833                 cfs_list_del_init(&mc->mc_linkage);
834                 lu_ref_del(&ld_c->ld_reference, "mdc-child", ld_m);
835                 lu_device_put(ld_c);
836                 ld_m->ld_type->ldt_ops->ldto_device_fini(env, ld_m);
837                 ld_m->ld_type->ldt_ops->ldto_device_free(env, ld_m);
838                 cm->cmm_tgt_count--;
839         }
840         cfs_spin_unlock(&cm->cmm_tgt_guard);
841
842         fld_client_proc_fini(cm->cmm_fld);
843         fld_client_fini(cm->cmm_fld);
844         ls = cmm2lu_dev(cm)->ld_site;
845         lu_site2md(ls)->ms_client_fld = NULL;
846         cmm_procfs_fini(cm);
847
848         RETURN (md2lu_dev(cm->cmm_child));
849 }
850
851 static struct lu_device_type_operations cmm_device_type_ops = {
852         .ldto_init = __cmm_type_init,
853         .ldto_fini = __cmm_type_fini,
854
855         .ldto_start = __cmm_type_start,
856         .ldto_stop  = __cmm_type_stop,
857
858         .ldto_device_alloc = cmm_device_alloc,
859         .ldto_device_free  = cmm_device_free,
860
861         .ldto_device_init = cmm_device_init,
862         .ldto_device_fini = cmm_device_fini
863 };
864
865 static struct lu_device_type cmm_device_type = {
866         .ldt_tags     = LU_DEVICE_MD,
867         .ldt_name     = LUSTRE_CMM_NAME,
868         .ldt_ops      = &cmm_device_type_ops,
869         .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD
870 };
871
872 struct lprocfs_vars lprocfs_cmm_obd_vars[] = {
873         { 0 }
874 };
875
876 struct lprocfs_vars lprocfs_cmm_module_vars[] = {
877         { 0 }
878 };
879
880 static void lprocfs_cmm_init_vars(struct lprocfs_static_vars *lvars)
881 {
882     lvars->module_vars  = lprocfs_cmm_module_vars;
883     lvars->obd_vars     = lprocfs_cmm_obd_vars;
884 }
885 /** @} */
886
887 static int __init cmm_mod_init(void)
888 {
889         struct lprocfs_static_vars lvars;
890
891         lprocfs_cmm_init_vars(&lvars);
892         return class_register_type(&cmm_obd_device_ops, NULL, lvars.module_vars,
893                                    LUSTRE_CMM_NAME, &cmm_device_type);
894 }
895
896 static void __exit cmm_mod_exit(void)
897 {
898         class_unregister_type(LUSTRE_CMM_NAME);
899 }
900
901 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
902 MODULE_DESCRIPTION("Lustre Clustered Metadata Manager ("LUSTRE_CMM_NAME")");
903 MODULE_LICENSE("GPL");
904
905 cfs_module(cmm, "0.1.0", cmm_mod_init, cmm_mod_exit);