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                       cfs_kstatfs_t *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);