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  *  lustre/cmm/cmm_device.c
   5  *  Lustre Cluster Metadata Manager (cmm)
   6  *
   7  *  Copyright (c) 2006 Cluster File Systems, Inc.
   8  *   Author: Mike Pershin <tappro@clusterfs.com>
   9  *
  10  *   This file is part of the Lustre file system, http://www.lustre.org
  11  *   Lustre is a trademark of Cluster File Systems, Inc.
  12  *
  13  *   You may have signed or agreed to another license before downloading
  14  *   this software.  If so, you are bound by the terms and conditions
  15  *   of that agreement, and the following does not apply to you.  See the
  16  *   LICENSE file included with this distribution for more information.
  17  *
  18  *   If you did not agree to a different license, then this copy of Lustre
  19  *   is open source software; you can redistribute it and/or modify it
  20  *   under the terms of version 2 of the GNU General Public License as
  21  *   published by the Free Software Foundation.
  22  *
  23  *   In either case, Lustre is distributed in the hope that it will be
  24  *   useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  25  *   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  26  *   license text for more details.
  27  */
  28
  29 #ifndef EXPORT_SYMTAB
  30 # define EXPORT_SYMTAB
  31 #endif
  32 #define DEBUG_SUBSYSTEM S_MDS
  33
  34 #include <linux/module.h>
  35
  36 #include <obd.h>
  37 #include <obd_class.h>
  38 #include <lprocfs_status.h>
  39 #include <lustre_ver.h>
  40 #include "cmm_internal.h"
  41 #include "mdc_internal.h"
  42
  43 static struct obd_ops cmm_obd_device_ops = {
  44         .o_owner           = THIS_MODULE
  45 };
  46
  47 static struct lu_device_operations cmm_lu_ops;
  48
  49 static inline int lu_device_is_cmm(struct lu_device *d)
  50 {
  51         return ergo(d != NULL && d->ld_ops != NULL, d->ld_ops == &cmm_lu_ops);
  52 }
  53
  54 int cmm_root_get(const struct lu_env *env, struct md_device *md,
  55                  struct lu_fid *fid)
  56 {
  57         struct cmm_device *cmm_dev = md2cmm_dev(md);
  58         /* valid only on master MDS */
  59         if (cmm_dev->cmm_local_num == 0)
  60                 return cmm_child_ops(cmm_dev)->mdo_root_get(env,
  61                                      cmm_dev->cmm_child, fid);
  62         else
  63                 return -EINVAL;
  64 }
  65
  66 static int cmm_statfs(const struct lu_env *env, struct md_device *md,
  67                       struct kstatfs *sfs)
  68 {
  69         struct cmm_device *cmm_dev = md2cmm_dev(md);
  70         int rc;
  71
  72         ENTRY;
  73         rc = cmm_child_ops(cmm_dev)->mdo_statfs(env,
  74                                                 cmm_dev->cmm_child, sfs);
  75         RETURN (rc);
  76 }
  77
  78 static int cmm_maxsize_get(const struct lu_env *env, struct md_device *md,
  79                            int *md_size, int *cookie_size)
  80 {
  81         struct cmm_device *cmm_dev = md2cmm_dev(md);
  82         int rc;
  83         ENTRY;
  84         rc = cmm_child_ops(cmm_dev)->mdo_maxsize_get(env, cmm_dev->cmm_child,
  85                                                      md_size, cookie_size);
  86         RETURN(rc);
  87 }
  88
  89 static int cmm_init_capa_keys(struct md_device *md,
  90                               struct lustre_capa_key *keys)
  91 {
  92         struct cmm_device *cmm_dev = md2cmm_dev(md);
  93         int rc;
  94         ENTRY;
  95         LASSERT(cmm_child_ops(cmm_dev)->mdo_init_capa_keys);
  96         rc = cmm_child_ops(cmm_dev)->mdo_init_capa_keys(cmm_dev->cmm_child,
  97                                                         keys);
  98         RETURN(rc);
  99 }
 100
 101 static int cmm_update_capa_key(const struct lu_env *env,
 102                                struct md_device *md,
 103                                struct lustre_capa_key *key)
 104 {
 105         struct cmm_device *cmm_dev = md2cmm_dev(md);
 106         int rc;
 107         ENTRY;
 108         rc = cmm_child_ops(cmm_dev)->mdo_update_capa_key(env,
 109                                                          cmm_dev->cmm_child,
 110                                                          key);
 111         RETURN(rc);
 112 }
 113
 114 static struct md_device_operations cmm_md_ops = {
 115         .mdo_statfs         = cmm_statfs,
 116         .mdo_root_get       = cmm_root_get,
 117         .mdo_maxsize_get    = cmm_maxsize_get,
 118         .mdo_init_capa_keys = cmm_init_capa_keys,
 119         .mdo_update_capa_key= cmm_update_capa_key,
 120 };
 121
 122 extern struct lu_device_type mdc_device_type;
 123
 124 /* --- cmm_lu_operations --- */
 125 /* add new MDC to the CMM, create MDC lu_device and connect it to mdc_obd */
 126 static int cmm_add_mdc(const struct lu_env *env,
 127                        struct cmm_device *cm, struct lustre_cfg *cfg)
 128 {
 129         struct lu_device_type *ldt = &mdc_device_type;
 130         char *p, *num = lustre_cfg_string(cfg, 2);
 131         struct mdc_device *mc, *tmp;
 132         struct lu_fld_target target;
 133         struct lu_device *ld;
 134         struct lu_site *ls;
 135         mdsno_t mdc_num;
 136         int rc;
 137         ENTRY;
 138
 139         /* find out that there is no such mdc */
 140         LASSERT(num);
 141         mdc_num = simple_strtol(num, &p, 10);
 142         if (*p) {
 143                 CERROR("Invalid index in lustre_cgf, offset 2\n");
 144                 RETURN(-EINVAL);
 145         }
 146
 147         spin_lock(&cm->cmm_tgt_guard);
 148         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
 149                                  mc_linkage) {
 150                 if (mc->mc_num == mdc_num) {
 151                         spin_unlock(&cm->cmm_tgt_guard);
 152                         RETURN(-EEXIST);
 153                 }
 154         }
 155         spin_unlock(&cm->cmm_tgt_guard);
 156         ld = ldt->ldt_ops->ldto_device_alloc(env, ldt, cfg);
 157         ld->ld_site = cmm2lu_dev(cm)->ld_site;
 158
 159         rc = ldt->ldt_ops->ldto_device_init(env, ld, NULL);
 160         if (rc) {
 161                 ldt->ldt_ops->ldto_device_free(env, ld);
 162                 RETURN (rc);
 163         }
 164         /* pass config to the just created MDC */
 165         rc = ld->ld_ops->ldo_process_config(env, ld, cfg);
 166         if (rc == 0) {
 167                 spin_lock(&cm->cmm_tgt_guard);
 168                 list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
 169                                          mc_linkage) {
 170                         if (mc->mc_num == mdc_num) {
 171                                 spin_unlock(&cm->cmm_tgt_guard);
 172                                 ldt->ldt_ops->ldto_device_fini(env, ld);
 173                                 ldt->ldt_ops->ldto_device_free(env, ld);
 174                                 RETURN(-EEXIST);
 175                         }
 176                 }
 177                 mc = lu2mdc_dev(ld);
 178                 list_add_tail(&mc->mc_linkage, &cm->cmm_targets);
 179                 cm->cmm_tgt_count++;
 180                 spin_unlock(&cm->cmm_tgt_guard);
 181
 182                 lu_device_get(cmm2lu_dev(cm));
 183
 184                 ls = cm->cmm_md_dev.md_lu_dev.ld_site;
 185
 186                 target.ft_srv = NULL;
 187                 target.ft_idx = mc->mc_num;
 188                 target.ft_exp = mc->mc_desc.cl_exp;
 189
 190                 fld_client_add_target(ls->ls_client_fld, &target);
 191         }
 192         RETURN(rc);
 193 }
 194
 195 static void cmm_device_shutdown(const struct lu_env *env,
 196                                 struct cmm_device *cm)
 197 {
 198         struct mdc_device *mc, *tmp;
 199         ENTRY;
 200
 201         /* finish all mdc devices */
 202         spin_lock(&cm->cmm_tgt_guard);
 203         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets, mc_linkage) {
 204                 struct lu_device *ld_m = mdc2lu_dev(mc);
 205
 206                 list_del_init(&mc->mc_linkage);
 207                 lu_device_put(cmm2lu_dev(cm));
 208                 ld_m->ld_type->ldt_ops->ldto_device_fini(env, ld_m);
 209                 ld_m->ld_type->ldt_ops->ldto_device_free(env, ld_m);
 210                 cm->cmm_tgt_count--;
 211         }
 212         spin_unlock(&cm->cmm_tgt_guard);
 213
 214         EXIT;
 215 }
 216 static int cmm_device_mount(const struct lu_env *env,
 217                             struct cmm_device *m, struct lustre_cfg *cfg)
 218 {
 219         const char *index = lustre_cfg_string(cfg, 2);
 220         char *p;
 221
 222         LASSERT(index != NULL);
 223
 224         m->cmm_local_num = simple_strtol(index, &p, 10);
 225         if (*p) {
 226                 CERROR("Invalid index in lustre_cgf\n");
 227                 RETURN(-EINVAL);
 228         }
 229
 230         RETURN(0);
 231 }
 232
 233 static int cmm_process_config(const struct lu_env *env,
 234                               struct lu_device *d, struct lustre_cfg *cfg)
 235 {
 236         struct cmm_device *m = lu2cmm_dev(d);
 237         struct lu_device *next = md2lu_dev(m->cmm_child);
 238         int err;
 239         ENTRY;
 240
 241         switch(cfg->lcfg_command) {
 242         case LCFG_ADD_MDC:
 243                 err = cmm_add_mdc(env, m, cfg);
 244                 /* the first ADD_MDC can be counted as setup is finished */
 245                 if ((m->cmm_flags & CMM_INITIALIZED) == 0)
 246                         m->cmm_flags |= CMM_INITIALIZED;
 247                 break;
 248         case LCFG_SETUP:
 249         {
 250                 /* lower layers should be set up at first */
 251                 err = next->ld_ops->ldo_process_config(env, next, cfg);
 252                 if (err == 0)
 253                         err = cmm_device_mount(env, m, cfg);
 254                 break;
 255         }
 256         case LCFG_CLEANUP:
 257         {
 258                 cmm_device_shutdown(env, m);
 259         }
 260         default:
 261                 err = next->ld_ops->ldo_process_config(env, next, cfg);
 262         }
 263         RETURN(err);
 264 }
 265
 266 static int cmm_recovery_complete(const struct lu_env *env,
 267                                  struct lu_device *d)
 268 {
 269         struct cmm_device *m = lu2cmm_dev(d);
 270         struct lu_device *next = md2lu_dev(m->cmm_child);
 271         int rc;
 272         ENTRY;
 273         rc = next->ld_ops->ldo_recovery_complete(env, next);
 274         RETURN(rc);
 275 }
 276
 277 static struct lu_device_operations cmm_lu_ops = {
 278         .ldo_object_alloc      = cmm_object_alloc,
 279         .ldo_process_config    = cmm_process_config,
 280         .ldo_recovery_complete = cmm_recovery_complete
 281 };
 282
 283 /* --- lu_device_type operations --- */
 284 int cmm_upcall(const struct lu_env *env, struct md_device *md,
 285                enum md_upcall_event ev)
 286 {
 287         struct md_device *upcall_dev;
 288         int rc;
 289         ENTRY;
 290
 291         upcall_dev = md->md_upcall.mu_upcall_dev;
 292
 293         LASSERT(upcall_dev);
 294         rc = upcall_dev->md_upcall.mu_upcall(env, md->md_upcall.mu_upcall_dev, ev);
 295
 296         RETURN(rc);
 297 }
 298
 299 static struct lu_device *cmm_device_alloc(const struct lu_env *env,
 300                                           struct lu_device_type *t,
 301                                           struct lustre_cfg *cfg)
 302 {
 303         struct lu_device  *l;
 304         struct cmm_device *m;
 305
 306         ENTRY;
 307
 308         OBD_ALLOC_PTR(m);
 309         if (m == NULL) {
 310                 l = ERR_PTR(-ENOMEM);
 311         } else {
 312                 md_device_init(&m->cmm_md_dev, t);
 313                 m->cmm_md_dev.md_ops = &cmm_md_ops;
 314                 m->cmm_md_dev.md_upcall.mu_upcall = cmm_upcall;
 315                 l = cmm2lu_dev(m);
 316                 l->ld_ops = &cmm_lu_ops;
 317         }
 318
 319         RETURN (l);
 320 }
 321
 322 static void cmm_device_free(const struct lu_env *env, struct lu_device *d)
 323 {
 324         struct cmm_device *m = lu2cmm_dev(d);
 325
 326         LASSERT(m->cmm_tgt_count == 0);
 327         LASSERT(list_empty(&m->cmm_targets));
 328         md_device_fini(&m->cmm_md_dev);
 329         OBD_FREE_PTR(m);
 330 }
 331
 332 /* context key constructor/destructor */
 333 static void *cmm_key_init(const struct lu_context *ctx,
 334                           struct lu_context_key *key)
 335 {
 336         struct cmm_thread_info *info;
 337
 338         CLASSERT(CFS_PAGE_SIZE >= sizeof *info);
 339         OBD_ALLOC_PTR(info);
 340         if (info == NULL)
 341                 info = ERR_PTR(-ENOMEM);
 342         return info;
 343 }
 344
 345 static void cmm_key_fini(const struct lu_context *ctx,
 346                          struct lu_context_key *key, void *data)
 347 {
 348         struct cmm_thread_info *info = data;
 349         OBD_FREE_PTR(info);
 350 }
 351
 352 struct lu_context_key cmm_thread_key = {
 353         .lct_tags = LCT_MD_THREAD,
 354         .lct_init = cmm_key_init,
 355         .lct_fini = cmm_key_fini
 356 };
 357
 358 static int cmm_type_init(struct lu_device_type *t)
 359 {
 360         return lu_context_key_register(&cmm_thread_key);
 361 }
 362
 363 static void cmm_type_fini(struct lu_device_type *t)
 364 {
 365         lu_context_key_degister(&cmm_thread_key);
 366 }
 367
 368 static int cmm_device_init(const struct lu_env *env,
 369                            struct lu_device *d, struct lu_device *next)
 370 {
 371         struct cmm_device *m = lu2cmm_dev(d);
 372         int err = 0;
 373         ENTRY;
 374
 375         spin_lock_init(&m->cmm_tgt_guard);
 376         INIT_LIST_HEAD(&m->cmm_targets);
 377         m->cmm_tgt_count = 0;
 378         m->cmm_child = lu2md_dev(next);
 379
 380         RETURN(err);
 381 }
 382
 383 static struct lu_device *cmm_device_fini(const struct lu_env *env,
 384                                          struct lu_device *ld)
 385 {
 386         struct cmm_device *cm = lu2cmm_dev(ld);
 387         ENTRY;
 388         RETURN (md2lu_dev(cm->cmm_child));
 389 }
 390
 391 static struct lu_device_type_operations cmm_device_type_ops = {
 392         .ldto_init = cmm_type_init,
 393         .ldto_fini = cmm_type_fini,
 394
 395         .ldto_device_alloc = cmm_device_alloc,
 396         .ldto_device_free  = cmm_device_free,
 397
 398         .ldto_device_init = cmm_device_init,
 399         .ldto_device_fini = cmm_device_fini
 400 };
 401
 402 static struct lu_device_type cmm_device_type = {
 403         .ldt_tags     = LU_DEVICE_MD,
 404         .ldt_name     = LUSTRE_CMM_NAME,
 405         .ldt_ops      = &cmm_device_type_ops,
 406         .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD
 407 };
 408
 409 struct lprocfs_vars lprocfs_cmm_obd_vars[] = {
 410         { 0 }
 411 };
 412
 413 struct lprocfs_vars lprocfs_cmm_module_vars[] = {
 414         { 0 }
 415 };
 416
 417 LPROCFS_INIT_VARS(cmm, lprocfs_cmm_module_vars, lprocfs_cmm_obd_vars);
 418
 419 static int __init cmm_mod_init(void)
 420 {
 421         struct lprocfs_static_vars lvars;
 422
 423         printk(KERN_INFO "Lustre: Clustered Metadata Manager; info@clusterfs.com\n");
 424
 425         lprocfs_init_vars(cmm, &lvars);
 426         return class_register_type(&cmm_obd_device_ops, NULL, lvars.module_vars,
 427                                    LUSTRE_CMM_NAME, &cmm_device_type);
 428 }
 429
 430 static void __exit cmm_mod_exit(void)
 431 {
 432         class_unregister_type(LUSTRE_CMM_NAME);
 433 }
 434
 435 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 436 MODULE_DESCRIPTION("Lustre Clustered Metadata Manager ("LUSTRE_CMM_NAME")");
 437 MODULE_LICENSE("GPL");
 438
 439 cfs_module(cmm, "0.1.0", cmm_mod_init, cmm_mod_exit);