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