lustre/lod/lod_dev.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright  2009 Sun Microsystems, Inc. All rights reserved
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, Intel Corporation.
  27  *
  28  */
  29 /*
  30  * This file is part of Lustre, http://www.lustre.org/
  31  * Lustre is a trademark of Sun Microsystems, Inc.
  32  *
  33  * lustre/lod/lod_dev.c
  34  *
  35  * Lustre Logical Object Device
  36  *
  37  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  38  * Author: Mikhail Pershin <mike.pershin@intel.com>
  39  */
  40
  41 #ifndef EXPORT_SYMTAB
  42 # define EXPORT_SYMTAB
  43 #endif
  44 #define DEBUG_SUBSYSTEM S_MDS
  45
  46 #include <obd_class.h>
  47 #include <lustre_param.h>
  48
  49 #include "lod_internal.h"
  50
  51 /**
  52  * Lookup MDT/OST index \a tgt by FID \a fid.
  53  *
  54  * \param lod LOD to be lookup at.
  55  * \param fid FID of object to find MDT/OST.
  56  * \param tgt MDT/OST index to return.
  57  * \param flags indidcate the FID is on MDS or OST.
  58  **/
  59 int lod_fld_lookup(const struct lu_env *env, struct lod_device *lod,
  60                    const struct lu_fid *fid, __u32 *tgt, int flags)
  61 {
  62         struct lu_seq_range     range;
  63         struct lu_server_fld    *server_fld;
  64         int rc = 0;
  65         ENTRY;
  66
  67         LASSERTF(fid_is_sane(fid), "Invalid FID "DFID"\n", PFID(fid));
  68         if (fid_is_idif(fid)) {
  69                 *tgt = fid_idif_ost_idx(fid);
  70                 RETURN(rc);
  71         }
  72
  73         if (!lod->lod_initialized || !fid_is_norm(fid)) {
  74                 LASSERT(lu_site2seq(lod2lu_dev(lod)->ld_site) != NULL);
  75                 *tgt = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
  76                 RETURN(rc);
  77         }
  78
  79         server_fld = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_server_fld;
  80         range.lsr_flags = flags;
  81         rc = fld_server_lookup(env, server_fld, fid_seq(fid), &range);
  82         if (rc) {
  83                 CERROR("%s: Can't find tgt by seq "LPX64", rc %d\n",
  84                        lod2obd(lod)->obd_name, fid_seq(fid), rc);
  85                 RETURN(rc);
  86         }
  87
  88         *tgt = range.lsr_index;
  89
  90         CDEBUG(D_INFO, "LOD: got tgt %x for sequence: "
  91                LPX64"\n", *tgt, fid_seq(fid));
  92
  93         RETURN(rc);
  94 }
  95
  96 extern struct lu_object_operations lod_lu_obj_ops;
  97 extern struct dt_object_operations lod_obj_ops;
  98
  99 /* Slab for OSD object allocation */
 100 cfs_mem_cache_t *lod_object_kmem;
 101
 102 static struct lu_kmem_descr lod_caches[] = {
 103         {
 104                 .ckd_cache = &lod_object_kmem,
 105                 .ckd_name  = "lod_obj",
 106                 .ckd_size  = sizeof(struct lod_object)
 107         },
 108         {
 109                 .ckd_cache = NULL
 110         }
 111 };
 112
 113 static struct lu_device *lod_device_fini(const struct lu_env *env,
 114                                          struct lu_device *d);
 115
 116 struct lu_object *lod_object_alloc(const struct lu_env *env,
 117                                    const struct lu_object_header *hdr,
 118                                    struct lu_device *dev)
 119 {
 120         struct lu_object  *lu_obj;
 121         struct lod_object *lo;
 122
 123         OBD_SLAB_ALLOC_PTR_GFP(lo, lod_object_kmem, CFS_ALLOC_IO);
 124         if (lo == NULL)
 125                 return NULL;
 126
 127         lu_obj = lod2lu_obj(lo);
 128         dt_object_init(&lo->ldo_obj, NULL, dev);
 129         lo->ldo_obj.do_ops = &lod_obj_ops;
 130         lu_obj->lo_ops = &lod_lu_obj_ops;
 131
 132         return lu_obj;
 133 }
 134
 135 static int lod_process_config(const struct lu_env *env,
 136                               struct lu_device *dev,
 137                               struct lustre_cfg *lcfg)
 138 {
 139         struct lod_device *lod = lu2lod_dev(dev);
 140         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
 141         char              *arg1;
 142         int                rc, i;
 143         ENTRY;
 144
 145         switch(lcfg->lcfg_command) {
 146
 147         case LCFG_LOV_DEL_OBD:
 148         case LCFG_LOV_ADD_INA:
 149         case LCFG_LOV_ADD_OBD: {
 150                 __u32 index;
 151                 int gen;
 152                 /* lov_modify_tgts add  0:lov_mdsA  1:osp  2:0  3:1 */
 153                 arg1 = lustre_cfg_string(lcfg, 1);
 154
 155                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
 156                         GOTO(out, rc = -EINVAL);
 157                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
 158                         GOTO(out, rc = -EINVAL);
 159
 160                 if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD)
 161                         rc = lod_add_device(env, lod, arg1, index, gen, 1);
 162                 else if (lcfg->lcfg_command == LCFG_LOV_ADD_INA)
 163                         rc = lod_add_device(env, lod, arg1, index, gen, 0);
 164                 else
 165                         rc = lod_del_device(env, lod, arg1, index, gen);
 166
 167                 break;
 168         }
 169
 170         case LCFG_PARAM: {
 171                 struct lprocfs_static_vars  v = { 0 };
 172                 struct obd_device         *obd = lod2obd(lod);
 173
 174                 lprocfs_lod_init_vars(&v);
 175
 176                 rc = class_process_proc_param(PARAM_LOV, v.obd_vars, lcfg, obd);
 177                 if (rc > 0)
 178                         rc = 0;
 179                 GOTO(out, rc);
 180          }
 181
 182         case LCFG_CLEANUP:
 183                 lu_dev_del_linkage(dev->ld_site, dev);
 184                 lod_getref(lod);
 185                 lod_foreach_ost(lod, i) {
 186                         struct lod_ost_desc *ost;
 187                         ost = OST_TGT(lod, i);
 188                         LASSERT(ost && ost->ltd_ost);
 189                         next = &ost->ltd_ost->dd_lu_dev;
 190                         rc = next->ld_ops->ldo_process_config(env, next, lcfg);
 191                         if (rc)
 192                                 CERROR("%s: can't process %u: %d\n",
 193                                        lod2obd(lod)->obd_name,
 194                                        lcfg->lcfg_command, rc);
 195                 }
 196                 lod_putref(lod);
 197
 198                 /*
 199                  * do cleanup on underlying storage only when
 200                  * all OSPs are cleaned up, as they use that OSD as well
 201                  */
 202                 next = &lod->lod_child->dd_lu_dev;
 203                 rc = next->ld_ops->ldo_process_config(env, next, lcfg);
 204                 if (rc)
 205                         CERROR("%s: can't process %u: %d\n",
 206                                lod2obd(lod)->obd_name, lcfg->lcfg_command, rc);
 207
 208                 rc = obd_disconnect(lod->lod_child_exp);
 209                 if (rc)
 210                         CERROR("error in disconnect from storage: %d\n", rc);
 211                 break;
 212
 213         default:
 214                CERROR("%s: unknown command %u\n", lod2obd(lod)->obd_name,
 215                       lcfg->lcfg_command);
 216                rc = -EINVAL;
 217                break;
 218         }
 219
 220 out:
 221         RETURN(rc);
 222 }
 223
 224 static int lod_recovery_complete(const struct lu_env *env,
 225                                  struct lu_device *dev)
 226 {
 227         struct lod_device   *lod = lu2lod_dev(dev);
 228         struct lu_device    *next = &lod->lod_child->dd_lu_dev;
 229         struct lod_ost_desc *ost;
 230         int                  i, rc;
 231         ENTRY;
 232
 233         LASSERT(lod->lod_recovery_completed == 0);
 234         lod->lod_recovery_completed = 1;
 235
 236         rc = next->ld_ops->ldo_recovery_complete(env, next);
 237
 238         lod_getref(lod);
 239         lod_foreach_ost(lod, i) {
 240                 ost = OST_TGT(lod, i);
 241                 LASSERT(ost && ost->ltd_ost);
 242                 next = &ost->ltd_ost->dd_lu_dev;
 243                 rc = next->ld_ops->ldo_recovery_complete(env, next);
 244                 if (rc)
 245                         CERROR("%s: can't complete recovery on #%d: %d\n",
 246                                lod2obd(lod)->obd_name, i, rc);
 247         }
 248         lod_putref(lod);
 249
 250         RETURN(rc);
 251 }
 252
 253 static int lod_prepare(const struct lu_env *env, struct lu_device *pdev,
 254                        struct lu_device *cdev)
 255 {
 256         struct lod_device   *lod = lu2lod_dev(cdev);
 257         struct lu_device    *next = &lod->lod_child->dd_lu_dev;
 258         int                  rc;
 259         ENTRY;
 260
 261         rc = next->ld_ops->ldo_prepare(env, pdev, next);
 262         if (rc != 0) {
 263                 CERROR("%s: prepare bottom error: rc = %d\n",
 264                        lod2obd(lod)->obd_name, rc);
 265                 RETURN(rc);
 266         }
 267
 268         lod->lod_initialized = 1;
 269
 270         RETURN(rc);
 271 }
 272
 273 const struct lu_device_operations lod_lu_ops = {
 274         .ldo_object_alloc       = lod_object_alloc,
 275         .ldo_process_config     = lod_process_config,
 276         .ldo_recovery_complete  = lod_recovery_complete,
 277         .ldo_prepare            = lod_prepare,
 278 };
 279
 280 static int lod_root_get(const struct lu_env *env,
 281                         struct dt_device *dev, struct lu_fid *f)
 282 {
 283         return dt_root_get(env, dt2lod_dev(dev)->lod_child, f);
 284 }
 285
 286 static int lod_statfs(const struct lu_env *env,
 287                       struct dt_device *dev, struct obd_statfs *sfs)
 288 {
 289         return dt_statfs(env, dt2lod_dev(dev)->lod_child, sfs);
 290 }
 291
 292 static struct thandle *lod_trans_create(const struct lu_env *env,
 293                                         struct dt_device *dev)
 294 {
 295         return dt_trans_create(env, dt2lod_dev(dev)->lod_child);
 296 }
 297
 298 static int lod_trans_start(const struct lu_env *env, struct dt_device *dev,
 299                            struct thandle *th)
 300 {
 301         return dt_trans_start(env, dt2lod_dev(dev)->lod_child, th);
 302 }
 303
 304 static int lod_trans_stop(const struct lu_env *env, struct thandle *th)
 305 {
 306         /* XXX: we don't know next device, will be fixed with DNE */
 307         return dt_trans_stop(env, th->th_dev, th);
 308 }
 309
 310 static void lod_conf_get(const struct lu_env *env,
 311                          const struct dt_device *dev,
 312                          struct dt_device_param *param)
 313 {
 314         dt_conf_get(env, dt2lod_dev((struct dt_device *)dev)->lod_child, param);
 315 }
 316
 317 static int lod_sync(const struct lu_env *env, struct dt_device *dev)
 318 {
 319         struct lod_device   *lod = dt2lod_dev(dev);
 320         struct lod_ost_desc *ost;
 321         int                  rc = 0, i;
 322         ENTRY;
 323
 324         lod_getref(lod);
 325         lod_foreach_ost(lod, i) {
 326                 ost = OST_TGT(lod, i);
 327                 LASSERT(ost && ost->ltd_ost);
 328                 rc = dt_sync(env, ost->ltd_ost);
 329                 if (rc) {
 330                         CERROR("%s: can't sync %u: %d\n",
 331                                lod2obd(lod)->obd_name, i, rc);
 332                         break;
 333                 }
 334         }
 335         lod_putref(lod);
 336         if (rc == 0)
 337                 rc = dt_sync(env, lod->lod_child);
 338
 339         RETURN(rc);
 340 }
 341
 342 static int lod_ro(const struct lu_env *env, struct dt_device *dev)
 343 {
 344         return dt_ro(env, dt2lod_dev(dev)->lod_child);
 345 }
 346
 347 static int lod_commit_async(const struct lu_env *env, struct dt_device *dev)
 348 {
 349         return dt_commit_async(env, dt2lod_dev(dev)->lod_child);
 350 }
 351
 352 static int lod_init_capa_ctxt(const struct lu_env *env, struct dt_device *dev,
 353                               int mode, unsigned long timeout,
 354                               __u32 alg, struct lustre_capa_key *keys)
 355 {
 356         struct dt_device *next = dt2lod_dev(dev)->lod_child;
 357         return dt_init_capa_ctxt(env, next, mode, timeout, alg, keys);
 358 }
 359
 360 static const struct dt_device_operations lod_dt_ops = {
 361         .dt_root_get         = lod_root_get,
 362         .dt_statfs           = lod_statfs,
 363         .dt_trans_create     = lod_trans_create,
 364         .dt_trans_start      = lod_trans_start,
 365         .dt_trans_stop       = lod_trans_stop,
 366         .dt_conf_get         = lod_conf_get,
 367         .dt_sync             = lod_sync,
 368         .dt_ro               = lod_ro,
 369         .dt_commit_async     = lod_commit_async,
 370         .dt_init_capa_ctxt   = lod_init_capa_ctxt,
 371 };
 372
 373 static int lod_connect_to_osd(const struct lu_env *env, struct lod_device *lod,
 374                               struct lustre_cfg *cfg)
 375 {
 376         struct obd_connect_data *data = NULL;
 377         struct obd_device       *obd;
 378         char                    *nextdev = NULL, *p, *s;
 379         int                      rc, len = 0;
 380         ENTRY;
 381
 382         LASSERT(cfg);
 383         LASSERT(lod->lod_child_exp == NULL);
 384
 385         /* compatibility hack: we still use old config logs
 386          * which specify LOV, but we need to learn underlying
 387          * OSD device, which is supposed to be:
 388          *  <fsname>-MDTxxxx-osd
 389          *
 390          * 2.x MGS generates lines like the following:
 391          *   #03 (176)lov_setup 0:lustre-MDT0000-mdtlov  1:(struct lov_desc)
 392          * 1.8 MGS generates lines like the following:
 393          *   #03 (168)lov_setup 0:lustre-mdtlov  1:(struct lov_desc)
 394          *
 395          * we use "-MDT" to differentiate 2.x from 1.8 */
 396
 397         if ((p = lustre_cfg_string(cfg, 0)) && strstr(p, "-mdtlov")) {
 398                 len = strlen(p) + 1;
 399                 OBD_ALLOC(nextdev, len);
 400                 if (nextdev == NULL)
 401                         GOTO(out, rc = -ENOMEM);
 402
 403                 strcpy(nextdev, p);
 404                 s = strstr(nextdev, "-mdtlov");
 405                 if (unlikely(s == NULL)) {
 406                         CERROR("unable to parse device name %s\n",
 407                                lustre_cfg_string(cfg, 0));
 408                         GOTO(out, rc = -EINVAL);
 409                 }
 410
 411                 if (strstr(nextdev, "-MDT")) {
 412                         /* 2.x config */
 413                         strcpy(s, "-osd");
 414                 } else {
 415                         /* 1.8 config */
 416                         strcpy(s, "-MDT0000-osd");
 417                 }
 418         } else {
 419                 CERROR("unable to parse device name %s\n",
 420                        lustre_cfg_string(cfg, 0));
 421                 GOTO(out, rc = -EINVAL);
 422         }
 423
 424         OBD_ALLOC_PTR(data);
 425         if (data == NULL)
 426                 GOTO(out, rc = -ENOMEM);
 427
 428         obd = class_name2obd(nextdev);
 429         if (obd == NULL) {
 430                 CERROR("can not locate next device: %s\n", nextdev);
 431                 GOTO(out, rc = -ENOTCONN);
 432         }
 433
 434         data->ocd_connect_flags = OBD_CONNECT_VERSION;
 435         data->ocd_version = LUSTRE_VERSION_CODE;
 436
 437         rc = obd_connect(env, &lod->lod_child_exp, obd, &obd->obd_uuid,
 438                          data, NULL);
 439         if (rc) {
 440                 CERROR("cannot connect to next dev %s (%d)\n", nextdev, rc);
 441                 GOTO(out, rc);
 442         }
 443
 444         lod->lod_dt_dev.dd_lu_dev.ld_site =
 445                 lod->lod_child_exp->exp_obd->obd_lu_dev->ld_site;
 446         LASSERT(lod->lod_dt_dev.dd_lu_dev.ld_site);
 447         lod->lod_child = lu2dt_dev(lod->lod_child_exp->exp_obd->obd_lu_dev);
 448
 449 out:
 450         if (data)
 451                 OBD_FREE_PTR(data);
 452         if (nextdev)
 453                 OBD_FREE(nextdev, len);
 454         RETURN(rc);
 455 }
 456
 457 static int lod_init0(const struct lu_env *env, struct lod_device *lod,
 458                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
 459 {
 460         struct dt_device_param ddp;
 461         struct obd_device     *obd;
 462         int                    rc;
 463         ENTRY;
 464
 465         obd = class_name2obd(lustre_cfg_string(cfg, 0));
 466         if (obd == NULL) {
 467                 CERROR("Cannot find obd with name %s\n",
 468                        lustre_cfg_string(cfg, 0));
 469                 RETURN(-ENODEV);
 470         }
 471
 472         obd->obd_lu_dev = &lod->lod_dt_dev.dd_lu_dev;
 473         lod->lod_dt_dev.dd_lu_dev.ld_obd = obd;
 474         lod->lod_dt_dev.dd_lu_dev.ld_ops = &lod_lu_ops;
 475         lod->lod_dt_dev.dd_ops = &lod_dt_ops;
 476
 477         rc = lod_connect_to_osd(env, lod, cfg);
 478         if (rc)
 479                 RETURN(rc);
 480
 481         dt_conf_get(env, &lod->lod_dt_dev, &ddp);
 482         lod->lod_osd_max_easize = ddp.ddp_max_ea_size;
 483
 484         /* setup obd to be used with old lov code */
 485         rc = lod_pools_init(lod, cfg);
 486         if (rc)
 487                 GOTO(out_disconnect, rc);
 488
 489         rc = lod_procfs_init(lod);
 490         if (rc)
 491                 GOTO(out_pools, rc);
 492
 493         mutex_init(&lod->lod_mutex);
 494         init_rwsem(&lod->lod_rw_sem);
 495         spin_lock_init(&lod->lod_desc_lock);
 496
 497         RETURN(0);
 498
 499 out_pools:
 500         lod_pools_fini(lod);
 501 out_disconnect:
 502         obd_disconnect(lod->lod_child_exp);
 503         RETURN(rc);
 504 }
 505
 506 static struct lu_device *lod_device_free(const struct lu_env *env,
 507                                          struct lu_device *lu)
 508 {
 509         struct lod_device *lod = lu2lod_dev(lu);
 510         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
 511         ENTRY;
 512
 513         LASSERT(cfs_atomic_read(&lu->ld_ref) == 0);
 514         dt_device_fini(&lod->lod_dt_dev);
 515         OBD_FREE_PTR(lod);
 516         RETURN(next);
 517 }
 518
 519 static struct lu_device *lod_device_alloc(const struct lu_env *env,
 520                                           struct lu_device_type *type,
 521                                           struct lustre_cfg *lcfg)
 522 {
 523         struct lod_device *lod;
 524         struct lu_device  *lu_dev;
 525
 526         OBD_ALLOC_PTR(lod);
 527         if (lod == NULL) {
 528                 lu_dev = ERR_PTR(-ENOMEM);
 529         } else {
 530                 int rc;
 531
 532                 lu_dev = lod2lu_dev(lod);
 533                 dt_device_init(&lod->lod_dt_dev, type);
 534                 rc = lod_init0(env, lod, type, lcfg);
 535                 if (rc != 0) {
 536                         lod_device_free(env, lu_dev);
 537                         lu_dev = ERR_PTR(rc);
 538                 }
 539         }
 540
 541         return lu_dev;
 542 }
 543
 544 static struct lu_device *lod_device_fini(const struct lu_env *env,
 545                                          struct lu_device *d)
 546 {
 547         struct lod_device *lod = lu2lod_dev(d);
 548         ENTRY;
 549
 550         lod_pools_fini(lod);
 551
 552         lod_procfs_fini(lod);
 553
 554         RETURN(NULL);
 555 }
 556
 557 /*
 558  * we use exports to track all LOD users
 559  */
 560 static int lod_obd_connect(const struct lu_env *env, struct obd_export **exp,
 561                            struct obd_device *obd, struct obd_uuid *cluuid,
 562                            struct obd_connect_data *data, void *localdata)
 563 {
 564         struct lod_device    *lod = lu2lod_dev(obd->obd_lu_dev);
 565         struct lustre_handle  conn;
 566         int                   rc;
 567         ENTRY;
 568
 569         CDEBUG(D_CONFIG, "connect #%d\n", lod->lod_connects);
 570
 571         rc = class_connect(&conn, obd, cluuid);
 572         if (rc)
 573                 RETURN(rc);
 574
 575         *exp = class_conn2export(&conn);
 576
 577         mutex_lock(&lod->lod_mutex);
 578         lod->lod_connects++;
 579         /* at the moment we expect the only user */
 580         LASSERT(lod->lod_connects == 1);
 581         mutex_unlock(&lod->lod_mutex);
 582
 583         RETURN(0);
 584 }
 585
 586 /*
 587  * once last export (we don't count self-export) disappeared
 588  * lod can be released
 589  */
 590 static int lod_obd_disconnect(struct obd_export *exp)
 591 {
 592         struct obd_device *obd = exp->exp_obd;
 593         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
 594         int                rc, release = 0;
 595         ENTRY;
 596
 597         /* Only disconnect the underlying layers on the final disconnect. */
 598         mutex_lock(&lod->lod_mutex);
 599         lod->lod_connects--;
 600         if (lod->lod_connects != 0) {
 601                 /* why should there be more than 1 connect? */
 602                 mutex_unlock(&lod->lod_mutex);
 603                 CERROR("%s: disconnect #%d\n", exp->exp_obd->obd_name,
 604                        lod->lod_connects);
 605                 goto out;
 606         }
 607         mutex_unlock(&lod->lod_mutex);
 608
 609         /* the last user of lod has gone, let's release the device */
 610         release = 1;
 611
 612 out:
 613         rc = class_disconnect(exp); /* bz 9811 */
 614
 615         if (rc == 0 && release)
 616                 class_manual_cleanup(obd);
 617         RETURN(rc);
 618 }
 619
 620 LU_KEY_INIT(lod, struct lod_thread_info);
 621
 622 static void lod_key_fini(const struct lu_context *ctx,
 623                 struct lu_context_key *key, void *data)
 624 {
 625         struct lod_thread_info *info = data;
 626         /* allocated in lod_get_lov_ea
 627          * XXX: this is overload, a tread may have such store but used only
 628          * once. Probably better would be pool of such stores per LOD.
 629          */
 630         if (info->lti_ea_store) {
 631                 OBD_FREE_LARGE(info->lti_ea_store, info->lti_ea_store_size);
 632                 info->lti_ea_store = NULL;
 633                 info->lti_ea_store_size = 0;
 634         }
 635         OBD_FREE_PTR(info);
 636 }
 637
 638 /* context key: lod_thread_key */
 639 LU_CONTEXT_KEY_DEFINE(lod, LCT_MD_THREAD);
 640
 641 LU_TYPE_INIT_FINI(lod, &lod_thread_key);
 642
 643 static struct lu_device_type_operations lod_device_type_ops = {
 644         .ldto_init           = lod_type_init,
 645         .ldto_fini           = lod_type_fini,
 646
 647         .ldto_start          = lod_type_start,
 648         .ldto_stop           = lod_type_stop,
 649
 650         .ldto_device_alloc   = lod_device_alloc,
 651         .ldto_device_free    = lod_device_free,
 652
 653         .ldto_device_fini    = lod_device_fini
 654 };
 655
 656 static struct lu_device_type lod_device_type = {
 657         .ldt_tags     = LU_DEVICE_DT,
 658         .ldt_name     = LUSTRE_LOD_NAME,
 659         .ldt_ops      = &lod_device_type_ops,
 660         .ldt_ctx_tags = LCT_MD_THREAD,
 661 };
 662
 663 static int lod_obd_health_check(const struct lu_env *env,
 664                 struct obd_device *obd)
 665 {
 666         struct lod_device   *d = lu2lod_dev(obd->obd_lu_dev);
 667         struct lod_ost_desc *ost;
 668         int                  i, rc = 1;
 669         ENTRY;
 670
 671         LASSERT(d);
 672         lod_getref(d);
 673         lod_foreach_ost(d, i) {
 674                 ost = OST_TGT(d, i);
 675                 LASSERT(ost && ost->ltd_ost);
 676                 rc = obd_health_check(env, ost->ltd_exp->exp_obd);
 677                 /* one healthy device is enough */
 678                 if (rc == 0)
 679                         break;
 680         }
 681         lod_putref(d);
 682         RETURN(rc);
 683 }
 684
 685 static struct obd_ops lod_obd_device_ops = {
 686         .o_owner        = THIS_MODULE,
 687         .o_connect      = lod_obd_connect,
 688         .o_disconnect   = lod_obd_disconnect,
 689         .o_health_check = lod_obd_health_check,
 690         .o_pool_new     = lod_pool_new,
 691         .o_pool_rem     = lod_pool_remove,
 692         .o_pool_add     = lod_pool_add,
 693         .o_pool_del     = lod_pool_del,
 694 };
 695
 696 static int __init lod_mod_init(void)
 697 {
 698         struct lprocfs_static_vars  lvars = { 0 };
 699         cfs_proc_dir_entry_t       *lov_proc_dir;
 700         int                         rc;
 701
 702         rc = lu_kmem_init(lod_caches);
 703         if (rc)
 704                 return rc;
 705
 706         lprocfs_lod_init_vars(&lvars);
 707
 708         rc = class_register_type(&lod_obd_device_ops, NULL, lvars.module_vars,
 709                                  LUSTRE_LOD_NAME, &lod_device_type);
 710         if (rc) {
 711                 lu_kmem_fini(lod_caches);
 712                 return rc;
 713         }
 714
 715         /* create "lov" entry in procfs for compatibility purposes */
 716         lov_proc_dir = lprocfs_srch(proc_lustre_root, "lov");
 717         if (lov_proc_dir == NULL) {
 718                 lov_proc_dir = lprocfs_register("lov", proc_lustre_root,
 719                                                 NULL, NULL);
 720                 if (IS_ERR(lov_proc_dir))
 721                         CERROR("lod: can't create compat entry \"lov\": %d\n",
 722                                (int)PTR_ERR(lov_proc_dir));
 723         }
 724
 725         return rc;
 726 }
 727
 728 static void __exit lod_mod_exit(void)
 729 {
 730
 731         lprocfs_try_remove_proc_entry("lov", proc_lustre_root);
 732
 733         class_unregister_type(LUSTRE_LOD_NAME);
 734         lu_kmem_fini(lod_caches);
 735 }
 736
 737 MODULE_AUTHOR("Whamcloud, Inc. <http://www.whamcloud.com/>");
 738 MODULE_DESCRIPTION("Lustre Logical Object Device ("LUSTRE_LOD_NAME")");
 739 MODULE_LICENSE("GPL");
 740
 741 module_init(lod_mod_init);
 742 module_exit(lod_mod_exit);
 743