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_ctxt(const struct lu_env *env, struct md_device *md,
  90                               int mode , unsigned long timeout, __u32 alg,
  91                               struct lustre_capa_key *keys)
  92 {
  93         struct cmm_device *cmm_dev = md2cmm_dev(md);
  94         int rc;
  95         ENTRY;
  96         LASSERT(cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt);
  97         rc = cmm_child_ops(cmm_dev)->mdo_init_capa_ctxt(env, cmm_dev->cmm_child,
  98                                                         mode, timeout, alg,
  99                                                         keys);
 100         RETURN(rc);
 101 }
 102
 103 static int cmm_update_capa_key(const struct lu_env *env,
 104                                struct md_device *md,
 105                                struct lustre_capa_key *key)
 106 {
 107         struct cmm_device *cmm_dev = md2cmm_dev(md);
 108         int rc;
 109         ENTRY;
 110         rc = cmm_child_ops(cmm_dev)->mdo_update_capa_key(env,
 111                                                          cmm_dev->cmm_child,
 112                                                          key);
 113         RETURN(rc);
 114 }
 115
 116 static struct md_device_operations cmm_md_ops = {
 117         .mdo_statfs          = cmm_statfs,
 118         .mdo_root_get        = cmm_root_get,
 119         .mdo_maxsize_get     = cmm_maxsize_get,
 120         .mdo_init_capa_ctxt  = cmm_init_capa_ctxt,
 121         .mdo_update_capa_key = cmm_update_capa_key,
 122 };
 123
 124 extern struct lu_device_type mdc_device_type;
 125
 126 static int cmm_post_init_mdc(const struct lu_env *env,
 127                              struct cmm_device *cmm)
 128 {
 129         int max_mdsize, max_cookiesize, rc;
 130         struct mdc_device *mc, *tmp;
 131
 132         /* get the max mdsize and cookiesize from lower layer */
 133         rc = cmm_maxsize_get(env, &cmm->cmm_md_dev, &max_mdsize,
 134                                                 &max_cookiesize);
 135         if (rc)
 136                 RETURN(rc);
 137
 138         spin_lock(&cmm->cmm_tgt_guard);
 139         list_for_each_entry_safe(mc, tmp, &cmm->cmm_targets,
 140                                  mc_linkage) {
 141                 mdc_init_ea_size(env, mc, max_mdsize, max_cookiesize);
 142         }
 143         spin_unlock(&cmm->cmm_tgt_guard);
 144         RETURN(rc);
 145 }
 146
 147 /* --- cmm_lu_operations --- */
 148 /* add new MDC to the CMM, create MDC lu_device and connect it to mdc_obd */
 149 static int cmm_add_mdc(const struct lu_env *env,
 150                        struct cmm_device *cm, struct lustre_cfg *cfg)
 151 {
 152         struct lu_device_type *ldt = &mdc_device_type;
 153         char *p, *num = lustre_cfg_string(cfg, 2);
 154         struct mdc_device *mc, *tmp;
 155         struct lu_fld_target target;
 156         struct lu_device *ld;
 157         mdsno_t mdc_num;
 158         int rc;
 159         ENTRY;
 160
 161         /* find out that there is no such mdc */
 162         LASSERT(num);
 163         mdc_num = simple_strtol(num, &p, 10);
 164         if (*p) {
 165                 CERROR("Invalid index in lustre_cgf, offset 2\n");
 166                 RETURN(-EINVAL);
 167         }
 168
 169         spin_lock(&cm->cmm_tgt_guard);
 170         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
 171                                  mc_linkage) {
 172                 if (mc->mc_num == mdc_num) {
 173                         spin_unlock(&cm->cmm_tgt_guard);
 174                         RETURN(-EEXIST);
 175                 }
 176         }
 177         spin_unlock(&cm->cmm_tgt_guard);
 178         ld = ldt->ldt_ops->ldto_device_alloc(env, ldt, cfg);
 179         ld->ld_site = cmm2lu_dev(cm)->ld_site;
 180
 181         rc = ldt->ldt_ops->ldto_device_init(env, ld, NULL, NULL);
 182         if (rc) {
 183                 ldt->ldt_ops->ldto_device_free(env, ld);
 184                 RETURN (rc);
 185         }
 186         /* pass config to the just created MDC */
 187         rc = ld->ld_ops->ldo_process_config(env, ld, cfg);
 188         if (rc)
 189                 RETURN(rc);
 190
 191         spin_lock(&cm->cmm_tgt_guard);
 192         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
 193                                  mc_linkage) {
 194                 if (mc->mc_num == mdc_num) {
 195                         spin_unlock(&cm->cmm_tgt_guard);
 196                         ldt->ldt_ops->ldto_device_fini(env, ld);
 197                         ldt->ldt_ops->ldto_device_free(env, ld);
 198                         RETURN(-EEXIST);
 199                 }
 200         }
 201         mc = lu2mdc_dev(ld);
 202         list_add_tail(&mc->mc_linkage, &cm->cmm_targets);
 203         cm->cmm_tgt_count++;
 204         spin_unlock(&cm->cmm_tgt_guard);
 205
 206         lu_device_get(cmm2lu_dev(cm));
 207
 208         target.ft_srv = NULL;
 209         target.ft_idx = mc->mc_num;
 210         target.ft_exp = mc->mc_desc.cl_exp;
 211         fld_client_add_target(cm->cmm_fld, &target);
 212
 213         /* Set max md size for the mdc. */
 214         rc = cmm_post_init_mdc(env, cm);
 215         RETURN(rc);
 216 }
 217
 218 static void cmm_device_shutdown(const struct lu_env *env,
 219                                 struct cmm_device *cm,
 220                                 struct lustre_cfg *cfg)
 221 {
 222         struct mdc_device *mc, *tmp;
 223         ENTRY;
 224
 225         /* Remove local target from FLD. */
 226         fld_client_del_target(cm->cmm_fld, cm->cmm_local_num);
 227
 228         /* Finish all mdc devices. */
 229         spin_lock(&cm->cmm_tgt_guard);
 230         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets, mc_linkage) {
 231                 struct lu_device *ld_m = mdc2lu_dev(mc);
 232                 fld_client_del_target(cm->cmm_fld, mc->mc_num);
 233                 ld_m->ld_ops->ldo_process_config(env, ld_m, cfg);
 234         }
 235         spin_unlock(&cm->cmm_tgt_guard);
 236
 237         /* remove upcall device*/
 238         md_upcall_fini(&cm->cmm_md_dev);
 239
 240         EXIT;
 241 }
 242
 243 static int cmm_device_mount(const struct lu_env *env,
 244                             struct cmm_device *m, struct lustre_cfg *cfg)
 245 {
 246         const char *index = lustre_cfg_string(cfg, 2);
 247         char *p;
 248
 249         LASSERT(index != NULL);
 250
 251         m->cmm_local_num = simple_strtol(index, &p, 10);
 252         if (*p) {
 253                 CERROR("Invalid index in lustre_cgf\n");
 254                 RETURN(-EINVAL);
 255         }
 256
 257         RETURN(0);
 258 }
 259
 260 static int cmm_process_config(const struct lu_env *env,
 261                               struct lu_device *d, struct lustre_cfg *cfg)
 262 {
 263         struct cmm_device *m = lu2cmm_dev(d);
 264         struct lu_device *next = md2lu_dev(m->cmm_child);
 265         int err;
 266         ENTRY;
 267
 268         switch(cfg->lcfg_command) {
 269         case LCFG_ADD_MDC:
 270                 /* On first ADD_MDC add also local target. */
 271                 if (!(m->cmm_flags & CMM_INITIALIZED)) {
 272                         struct lu_site *ls = cmm2lu_dev(m)->ld_site;
 273                         struct lu_fld_target target;
 274
 275                         target.ft_srv = ls->ls_server_fld;
 276                         target.ft_idx = m->cmm_local_num;
 277                         target.ft_exp = NULL;
 278
 279                         fld_client_add_target(m->cmm_fld, &target);
 280                 }
 281                 err = cmm_add_mdc(env, m, cfg);
 282
 283                 /* The first ADD_MDC can be counted as setup is finished. */
 284                 if (!(m->cmm_flags & CMM_INITIALIZED))
 285                         m->cmm_flags |= CMM_INITIALIZED;
 286
 287                 break;
 288         case LCFG_SETUP:
 289         {
 290                 /* lower layers should be set up at first */
 291                 err = next->ld_ops->ldo_process_config(env, next, cfg);
 292                 if (err == 0)
 293                         err = cmm_device_mount(env, m, cfg);
 294                 break;
 295         }
 296         case LCFG_CLEANUP:
 297         {
 298                 cmm_device_shutdown(env, m, cfg);
 299         }
 300         default:
 301                 err = next->ld_ops->ldo_process_config(env, next, cfg);
 302         }
 303         RETURN(err);
 304 }
 305
 306 static int cmm_recovery_complete(const struct lu_env *env,
 307                                  struct lu_device *d)
 308 {
 309         struct cmm_device *m = lu2cmm_dev(d);
 310         struct lu_device *next = md2lu_dev(m->cmm_child);
 311         int rc;
 312         ENTRY;
 313         rc = next->ld_ops->ldo_recovery_complete(env, next);
 314         RETURN(rc);
 315 }
 316
 317 static struct lu_device_operations cmm_lu_ops = {
 318         .ldo_object_alloc      = cmm_object_alloc,
 319         .ldo_process_config    = cmm_process_config,
 320         .ldo_recovery_complete = cmm_recovery_complete
 321 };
 322
 323 /* --- lu_device_type operations --- */
 324 int cmm_upcall(const struct lu_env *env, struct md_device *md,
 325                enum md_upcall_event ev)
 326 {
 327         int rc;
 328         ENTRY;
 329
 330         switch (ev) {
 331                 case MD_LOV_SYNC:
 332                         rc = cmm_post_init_mdc(env, md2cmm_dev(md));
 333                         if (rc)
 334                                 CERROR("can not init md size %d\n", rc);
 335                         /* fall through */
 336                 default:
 337                         rc = md_do_upcall(env, md, ev);
 338         }
 339         RETURN(rc);
 340 }
 341
 342 static struct lu_device *cmm_device_alloc(const struct lu_env *env,
 343                                           struct lu_device_type *t,
 344                                           struct lustre_cfg *cfg)
 345 {
 346         struct lu_device  *l;
 347         struct cmm_device *m;
 348         ENTRY;
 349
 350         OBD_ALLOC_PTR(m);
 351         if (m == NULL) {
 352                 l = ERR_PTR(-ENOMEM);
 353         } else {
 354                 md_device_init(&m->cmm_md_dev, t);
 355                 m->cmm_md_dev.md_ops = &cmm_md_ops;
 356                 md_upcall_init(&m->cmm_md_dev, cmm_upcall);
 357                 l = cmm2lu_dev(m);
 358                 l->ld_ops = &cmm_lu_ops;
 359
 360                 OBD_ALLOC_PTR(m->cmm_fld);
 361                 if (!m->cmm_fld)
 362                         GOTO(out_free_cmm, l = ERR_PTR(-ENOMEM));
 363         }
 364
 365         RETURN(l);
 366 out_free_cmm:
 367         OBD_FREE_PTR(m);
 368         return l;
 369 }
 370
 371 static void cmm_device_free(const struct lu_env *env, struct lu_device *d)
 372 {
 373         struct cmm_device *m = lu2cmm_dev(d);
 374
 375         LASSERT(m->cmm_tgt_count == 0);
 376         LASSERT(list_empty(&m->cmm_targets));
 377         if (m->cmm_fld != NULL) {
 378                 OBD_FREE_PTR(m->cmm_fld);
 379                 m->cmm_fld = NULL;
 380         }
 381         md_device_fini(&m->cmm_md_dev);
 382         OBD_FREE_PTR(m);
 383 }
 384
 385 /* context key constructor/destructor */
 386 static void *cmm_key_init(const struct lu_context *ctx,
 387                           struct lu_context_key *key)
 388 {
 389         struct cmm_thread_info *info;
 390
 391         CLASSERT(CFS_PAGE_SIZE >= sizeof *info);
 392         OBD_ALLOC_PTR(info);
 393         if (info == NULL)
 394                 info = ERR_PTR(-ENOMEM);
 395         return info;
 396 }
 397
 398 static void cmm_key_fini(const struct lu_context *ctx,
 399                          struct lu_context_key *key, void *data)
 400 {
 401         struct cmm_thread_info *info = data;
 402         OBD_FREE_PTR(info);
 403 }
 404
 405 static struct lu_context_key cmm_thread_key = {
 406         .lct_tags = LCT_MD_THREAD,
 407         .lct_init = cmm_key_init,
 408         .lct_fini = cmm_key_fini
 409 };
 410
 411 struct cmm_thread_info *cmm_env_info(const struct lu_env *env)
 412 {
 413         struct cmm_thread_info *info;
 414
 415         info = lu_context_key_get(&env->le_ctx, &cmm_thread_key);
 416         LASSERT(info != NULL);
 417         return info;
 418 }
 419
 420 static int cmm_type_init(struct lu_device_type *t)
 421 {
 422         LU_CONTEXT_KEY_INIT(&cmm_thread_key);
 423         return lu_context_key_register(&cmm_thread_key);
 424 }
 425
 426 static void cmm_type_fini(struct lu_device_type *t)
 427 {
 428         lu_context_key_degister(&cmm_thread_key);
 429 }
 430
 431 static int cmm_device_init(const struct lu_env *env, struct lu_device *d,
 432                            const char *name, struct lu_device *next)
 433 {
 434         struct cmm_device *m = lu2cmm_dev(d);
 435         struct lu_site *ls;
 436         int err = 0;
 437         ENTRY;
 438
 439         spin_lock_init(&m->cmm_tgt_guard);
 440         INIT_LIST_HEAD(&m->cmm_targets);
 441         m->cmm_tgt_count = 0;
 442         m->cmm_child = lu2md_dev(next);
 443
 444         err = fld_client_init(m->cmm_fld, name,
 445                               LUSTRE_CLI_FLD_HASH_DHT);
 446         if (err) {
 447                 CERROR("Can't init FLD, err %d\n", err);
 448                 RETURN(err);
 449         }
 450
 451         /* Assign site's fld client ref, needed for asserts in osd. */
 452         ls = cmm2lu_dev(m)->ld_site;
 453         ls->ls_client_fld = m->cmm_fld;
 454         err = cmm_procfs_init(m, name);
 455
 456         RETURN(err);
 457 }
 458
 459 static struct lu_device *cmm_device_fini(const struct lu_env *env,
 460                                          struct lu_device *ld)
 461 {
 462         struct cmm_device *cm = lu2cmm_dev(ld);
 463         struct mdc_device *mc, *tmp;
 464         struct lu_site *ls;
 465         ENTRY;
 466
 467         /* Finish all mdc devices */
 468         spin_lock(&cm->cmm_tgt_guard);
 469         list_for_each_entry_safe(mc, tmp, &cm->cmm_targets, mc_linkage) {
 470                 struct lu_device *ld_m = mdc2lu_dev(mc);
 471
 472                 list_del_init(&mc->mc_linkage);
 473                 lu_device_put(cmm2lu_dev(cm));
 474                 ld_m->ld_type->ldt_ops->ldto_device_fini(env, ld_m);
 475                 ld_m->ld_type->ldt_ops->ldto_device_free(env, ld_m);
 476                 cm->cmm_tgt_count--;
 477         }
 478         spin_unlock(&cm->cmm_tgt_guard);
 479
 480         fld_client_fini(cm->cmm_fld);
 481         ls = cmm2lu_dev(cm)->ld_site;
 482         ls->ls_client_fld = NULL;
 483         cmm_procfs_fini(cm);
 484
 485         RETURN (md2lu_dev(cm->cmm_child));
 486 }
 487
 488 static struct lu_device_type_operations cmm_device_type_ops = {
 489         .ldto_init = cmm_type_init,
 490         .ldto_fini = cmm_type_fini,
 491
 492         .ldto_device_alloc = cmm_device_alloc,
 493         .ldto_device_free  = cmm_device_free,
 494
 495         .ldto_device_init = cmm_device_init,
 496         .ldto_device_fini = cmm_device_fini
 497 };
 498
 499 static struct lu_device_type cmm_device_type = {
 500         .ldt_tags     = LU_DEVICE_MD,
 501         .ldt_name     = LUSTRE_CMM_NAME,
 502         .ldt_ops      = &cmm_device_type_ops,
 503         .ldt_ctx_tags = LCT_MD_THREAD | LCT_DT_THREAD
 504 };
 505
 506 struct lprocfs_vars lprocfs_cmm_obd_vars[] = {
 507         { 0 }
 508 };
 509
 510 struct lprocfs_vars lprocfs_cmm_module_vars[] = {
 511         { 0 }
 512 };
 513
 514 LPROCFS_INIT_VARS(cmm, lprocfs_cmm_module_vars, lprocfs_cmm_obd_vars);
 515
 516 static int __init cmm_mod_init(void)
 517 {
 518         struct lprocfs_static_vars lvars;
 519
 520         lprocfs_init_vars(cmm, &lvars);
 521         return class_register_type(&cmm_obd_device_ops, NULL, lvars.module_vars,
 522                                    LUSTRE_CMM_NAME, &cmm_device_type);
 523 }
 524
 525 static void __exit cmm_mod_exit(void)
 526 {
 527         class_unregister_type(LUSTRE_CMM_NAME);
 528 }
 529
 530 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 531 MODULE_DESCRIPTION("Lustre Clustered Metadata Manager ("LUSTRE_CMM_NAME")");
 532 MODULE_LICENSE("GPL");
 533
 534 cfs_module(cmm, "0.1.0", cmm_mod_init, cmm_mod_exit);