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, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * lustre/lod/lod_dev.c
  33  *
  34  * Lustre Logical Object Device
  35  *
  36  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  37  * Author: Mikhail Pershin <mike.pershin@intel.com>
  38  */
  39 /**
  40  * The Logical Object Device (LOD) layer manages access to striped
  41  * objects (both regular files and directories). It implements the DT
  42  * device and object APIs and is responsible for creating, storing,
  43  * and loading striping information as an extended attribute of the
  44  * underlying OSD object. LOD is the server side analog of the LOV and
  45  * LMV layers on the client side.
  46  *
  47  * Metadata LU object stack (layers of the same compound LU object,
  48  * all have the same FID):
  49  *
  50  *        MDT
  51  *         |      MD API
  52  *        MDD
  53  *         |      DT API
  54  *        LOD
  55  *       /   \    DT API
  56  *     OSD   OSP
  57  *
  58  * During LOD object initialization the localness or remoteness of the
  59  * object FID dictates the choice between OSD and OSP.
  60  *
  61  * An LOD object (file or directory) with N stripes (each has a
  62  * different FID):
  63  *
  64  *          LOD
  65  *           |
  66  *   +---+---+---+...+
  67  *   |   |   |   |   |
  68  *   S0  S1  S2  S3  S(N-1)  OS[DP] objects, seen as DT objects by LOD
  69  *
  70  * When upper layers must access an object's stripes (which are
  71  * themselves OST or MDT LU objects) LOD finds these objects by their
  72  * FIDs and stores them as an array of DT object pointers on the
  73  * object. Declarations and operations on LOD objects are received by
  74  * LOD (as DT object operations) and performed on the underlying
  75  * OS[DP] object and (as needed) on the stripes. From the perspective
  76  * of LOD, a stripe-less file (created by mknod() or open with
  77  * O_LOV_DELAY_CREATE) is an object which does not yet have stripes,
  78  * while a non-striped directory (created by mkdir()) is an object
  79  * which will never have stripes.
  80  *
  81  * The LOD layer also implements a small subset of the OBD device API
  82  * to support MDT stack initialization and finalization (an MDD device
  83  * connects and disconnects itself to and from the underlying LOD
  84  * device), and pool management. In turn LOD uses the OBD device API
  85  * to connect it self to the underlying OSD, and to connect itself to
  86  * OSP devices representing the MDTs and OSTs that bear the stripes of
  87  * its objects.
  88  */
  89
  90 #define DEBUG_SUBSYSTEM S_MDS
  91
  92 #include <linux/kthread.h>
  93 #include <obd_class.h>
  94 #include <md_object.h>
  95 #include <lustre_fid.h>
  96 #include <uapi/linux/lustre/lustre_param.h>
  97 #include <lustre_update.h>
  98 #include <lustre_log.h>
  99
 100 #include "lod_internal.h"
 101
 102 static const char lod_update_log_name[] = "update_log";
 103 static const char lod_update_log_dir_name[] = "update_log_dir";
 104
 105 /*
 106  * Lookup target by FID.
 107  *
 108  * Lookup MDT/OST target index by FID. Type of the target can be
 109  * specific or any.
 110  *
 111  * \param[in] env               LU environment provided by the caller
 112  * \param[in] lod               lod device
 113  * \param[in] fid               FID
 114  * \param[out] tgt              result target index
 115  * \param[in] type              expected type of the target:
 116  *                              LU_SEQ_RANGE_{MDT,OST,ANY}
 117  *
 118  * \retval 0                    on success
 119  * \retval negative             negated errno on error
 120  **/
 121 int lod_fld_lookup(const struct lu_env *env, struct lod_device *lod,
 122                    const struct lu_fid *fid, u32 *tgt, int *type)
 123 {
 124         struct lu_seq_range range = { 0 };
 125         struct lu_server_fld *server_fld;
 126         int rc;
 127
 128         ENTRY;
 129
 130         if (!fid_is_sane(fid)) {
 131                 CERROR("%s: invalid FID "DFID"\n", lod2obd(lod)->obd_name,
 132                        PFID(fid));
 133                 RETURN(-EIO);
 134         }
 135
 136         if (fid_is_idif(fid)) {
 137                 *tgt = fid_idif_ost_idx(fid);
 138                 *type = LU_SEQ_RANGE_OST;
 139                 RETURN(0);
 140         }
 141
 142         if (fid_is_update_log(fid) || fid_is_update_log_dir(fid)) {
 143                 *tgt = fid_oid(fid);
 144                 *type = LU_SEQ_RANGE_MDT;
 145                 RETURN(0);
 146         }
 147
 148         if (!lod->lod_initialized || (!fid_seq_in_fldb(fid_seq(fid)))) {
 149                 LASSERT(lu_site2seq(lod2lu_dev(lod)->ld_site) != NULL);
 150
 151                 *tgt = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
 152                 *type = LU_SEQ_RANGE_MDT;
 153                 RETURN(0);
 154         }
 155
 156         server_fld = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_server_fld;
 157         if (!server_fld)
 158                 RETURN(-EIO);
 159
 160         fld_range_set_type(&range, *type);
 161         rc = fld_server_lookup(env, server_fld, fid_seq(fid), &range);
 162         if (rc != 0)
 163                 RETURN(rc);
 164
 165         *tgt = range.lsr_index;
 166         *type = range.lsr_flags;
 167
 168         CDEBUG(D_INFO, "%s: got tgt %x for sequence: %#llx\n",
 169                lod2obd(lod)->obd_name, *tgt, fid_seq(fid));
 170
 171         RETURN(0);
 172 }
 173
 174 /* Slab for OSD object allocation */
 175 struct kmem_cache *lod_object_kmem;
 176
 177 /* Slab for dt_txn_callback */
 178 struct kmem_cache *lod_txn_callback_kmem;
 179 static struct lu_kmem_descr lod_caches[] = {
 180         {
 181                 .ckd_cache = &lod_object_kmem,
 182                 .ckd_name  = "lod_obj",
 183                 .ckd_size  = sizeof(struct lod_object)
 184         },
 185         {
 186                 .ckd_cache = &lod_txn_callback_kmem,
 187                 .ckd_name  = "lod_txn_callback",
 188                 .ckd_size  = sizeof(struct dt_txn_callback)
 189         },
 190         {
 191                 .ckd_cache = NULL
 192         }
 193 };
 194
 195 static struct lu_device *lod_device_fini(const struct lu_env *env,
 196                                          struct lu_device *d);
 197
 198 /**
 199  * Implementation of lu_device_operations::ldo_object_alloc() for LOD
 200  *
 201  * Allocates and initializes LOD's slice in the given object.
 202  *
 203  * see include/lu_object.h for the details.
 204  */
 205 static struct lu_object *lod_object_alloc(const struct lu_env *env,
 206                                           const struct lu_object_header *hdr,
 207                                           struct lu_device *dev)
 208 {
 209         struct lod_object *lod_obj;
 210         struct lu_object *lu_obj;
 211
 212         ENTRY;
 213
 214         OBD_SLAB_ALLOC_PTR_GFP(lod_obj, lod_object_kmem, GFP_NOFS);
 215         if (!lod_obj)
 216                 RETURN(ERR_PTR(-ENOMEM));
 217
 218         mutex_init(&lod_obj->ldo_layout_mutex);
 219         lu_obj = lod2lu_obj(lod_obj);
 220         dt_object_init(&lod_obj->ldo_obj, NULL, dev);
 221         lod_obj->ldo_obj.do_ops = &lod_obj_ops;
 222         lu_obj->lo_ops = &lod_lu_obj_ops;
 223
 224         RETURN(lu_obj);
 225 }
 226
 227 /**
 228  * Process the config log for all sub device.
 229  *
 230  * The function goes through all the targets in the given table
 231  * and apply given configuration command on to the targets.
 232  * Used to cleanup the targets at unmount.
 233  *
 234  * \param[in] env               LU environment provided by the caller
 235  * \param[in] lod               lod device
 236  * \param[in] ltd               target's table to go through
 237  * \param[in] lcfg              configuration command to apply
 238  *
 239  * \retval 0                    on success
 240  * \retval negative             negated errno on error
 241  **/
 242 static int lod_sub_process_config(const struct lu_env *env,
 243                                  struct lod_device *lod,
 244                                  struct lod_tgt_descs *ltd,
 245                                  struct lustre_cfg *lcfg)
 246 {
 247         struct lu_device *next;
 248         struct lu_tgt_desc *tgt;
 249         int rc = 0;
 250
 251         lod_getref(ltd);
 252         ltd_foreach_tgt(ltd, tgt) {
 253                 int rc1;
 254
 255                 LASSERT(tgt && tgt->ltd_tgt);
 256                 next = &tgt->ltd_tgt->dd_lu_dev;
 257                 rc1 = next->ld_ops->ldo_process_config(env, next, lcfg);
 258                 if (rc1) {
 259                         CERROR("%s: error cleaning up LOD index %u: cmd %#x : rc = %d\n",
 260                                lod2obd(lod)->obd_name, tgt->ltd_index,
 261                                lcfg->lcfg_command, rc1);
 262                         rc = rc1;
 263                 }
 264         }
 265         lod_putref(lod, ltd);
 266         return rc;
 267 }
 268
 269 struct lod_recovery_data {
 270         struct lod_device       *lrd_lod;
 271         struct lod_tgt_desc     *lrd_ltd;
 272         struct task_struct      **lrd_task;
 273         u32                     lrd_idx;
 274         struct lu_env           lrd_env;
 275         struct completion       *lrd_started;
 276 };
 277
 278
 279 /**
 280  * process update recovery record
 281  *
 282  * Add the update recovery recode to the update recovery list in
 283  * lod_recovery_data. Then the recovery thread (target_recovery_thread)
 284  * will redo these updates.
 285  *
 286  * \param[in]env        execution environment
 287  * \param[in]llh        log handle of update record
 288  * \param[in]rec        update record to be replayed
 289  * \param[in]data       update recovery data which holds the necessary
 290  *                      arguments for recovery (see struct lod_recovery_data)
 291  *
 292  * \retval              0 if the record is processed successfully.
 293  * \retval              negative errno if the record processing fails.
 294  */
 295 static int lod_process_recovery_updates(const struct lu_env *env,
 296                                         struct llog_handle *llh,
 297                                         struct llog_rec_hdr *rec,
 298                                         void *data)
 299 {
 300         struct lod_recovery_data *lrd = data;
 301         struct llog_cookie *cookie = &lod_env_info(env)->lti_cookie;
 302         struct lu_target *lut;
 303         u32 index = 0;
 304
 305         ENTRY;
 306
 307         if (!lrd->lrd_ltd) {
 308                 int rc;
 309
 310                 rc = lodname2mdt_index(lod2obd(lrd->lrd_lod)->obd_name, &index);
 311                 if (rc != 0)
 312                         return rc;
 313         } else {
 314                 index = lrd->lrd_ltd->ltd_index;
 315         }
 316
 317         if (rec->lrh_len !=
 318                 llog_update_record_size((struct llog_update_record *)rec)) {
 319                 CERROR("%s: broken update record! index %u "DFID".%u: rc = %d\n",
 320                        lod2obd(lrd->lrd_lod)->obd_name, index,
 321                        PFID(&llh->lgh_id.lgl_oi.oi_fid), rec->lrh_index, -EIO);
 322                 return -EINVAL;
 323         }
 324
 325         cookie->lgc_lgl = llh->lgh_id;
 326         cookie->lgc_index = rec->lrh_index;
 327         cookie->lgc_subsys = LLOG_UPDATELOG_ORIG_CTXT;
 328
 329         CDEBUG(D_HA, "%s: process recovery updates "DFID".%u\n",
 330                lod2obd(lrd->lrd_lod)->obd_name,
 331                PFID(&llh->lgh_id.lgl_oi.oi_fid), rec->lrh_index);
 332         lut = lod2lu_dev(lrd->lrd_lod)->ld_site->ls_tgt;
 333
 334         if (lut->lut_obd->obd_stopping ||
 335             lut->lut_obd->obd_abort_recovery)
 336                 return -ESHUTDOWN;
 337
 338         return insert_update_records_to_replay_list(lut->lut_tdtd,
 339                                         (struct llog_update_record *)rec,
 340                                         cookie, index);
 341 }
 342
 343 /**
 344  * recovery thread for update log
 345  *
 346  * Start recovery thread and prepare the sub llog, then it will retrieve
 347  * the update records from the correpondent MDT and do recovery.
 348  *
 349  * \param[in] arg       pointer to the recovery data
 350  *
 351  * \retval              0 if recovery succeeds
 352  * \retval              negative errno if recovery failed.
 353  */
 354 static int lod_sub_recovery_thread(void *arg)
 355 {
 356         struct lod_recovery_data *lrd = arg;
 357         struct lod_device *lod = lrd->lrd_lod;
 358         struct dt_device *dt;
 359         struct llog_ctxt *ctxt = NULL;
 360         struct lu_env *env = &lrd->lrd_env;
 361         struct lu_target *lut;
 362         struct lu_tgt_desc *mdt = NULL;
 363         time64_t start;
 364         int retries = 0;
 365         int rc;
 366
 367         ENTRY;
 368
 369         lut = lod2lu_dev(lod)->ld_site->ls_tgt;
 370         atomic_inc(&lut->lut_tdtd->tdtd_recovery_threads_count);
 371         if (!lrd->lrd_ltd)
 372                 dt = lod->lod_child;
 373         else
 374                 dt = lrd->lrd_ltd->ltd_tgt;
 375
 376         start = ktime_get_real_seconds();
 377         complete(lrd->lrd_started);
 378
 379 again:
 380
 381         if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_TGT_RECOVERY_CONNECT)) &&
 382             lrd->lrd_ltd) {
 383                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_RECOVERY_CONNECT, cfs_fail_val);
 384                 rc = -EIO;
 385         } else {
 386                 rc = lod_sub_prep_llog(env, lod, dt, lrd->lrd_idx);
 387         }
 388         if (!rc && !lod->lod_child->dd_rdonly) {
 389                 /* Process the recovery record */
 390                 ctxt = llog_get_context(dt->dd_lu_dev.ld_obd,
 391                                         LLOG_UPDATELOG_ORIG_CTXT);
 392                 LASSERT(ctxt != NULL);
 393                 LASSERT(ctxt->loc_handle != NULL);
 394
 395                 rc = llog_cat_process(env, ctxt->loc_handle,
 396                                       lod_process_recovery_updates, lrd, 0, 0);
 397         }
 398
 399         if (rc < 0) {
 400                 struct lu_device *top_device;
 401
 402                 top_device = lod->lod_dt_dev.dd_lu_dev.ld_site->ls_top_dev;
 403                 /*
 404                  * Because the remote target might failover at the same time,
 405                  * let's retry here
 406                  */
 407                 if ((rc == -ETIMEDOUT || rc == -EAGAIN || rc == -EIO) &&
 408                      dt != lod->lod_child &&
 409                     !top_device->ld_obd->obd_abort_recovery &&
 410                     !top_device->ld_obd->obd_stopping) {
 411                         if (ctxt) {
 412                                 if (ctxt->loc_handle)
 413                                         llog_cat_close(env,
 414                                                        ctxt->loc_handle);
 415                                 llog_ctxt_put(ctxt);
 416                         }
 417                         retries++;
 418                         CDEBUG(D_HA, "%s get update log failed %d, retry\n",
 419                                dt->dd_lu_dev.ld_obd->obd_name, rc);
 420                         goto again;
 421                 }
 422
 423                 CERROR("%s get update log failed: rc = %d\n",
 424                        dt->dd_lu_dev.ld_obd->obd_name, rc);
 425                 llog_ctxt_put(ctxt);
 426
 427                 spin_lock(&top_device->ld_obd->obd_dev_lock);
 428                 if (!top_device->ld_obd->obd_abort_recovery &&
 429                     !top_device->ld_obd->obd_stopping)
 430                         top_device->ld_obd->obd_abort_recovery = 1;
 431                 spin_unlock(&top_device->ld_obd->obd_dev_lock);
 432
 433                 GOTO(out, rc);
 434         }
 435         llog_ctxt_put(ctxt);
 436
 437         CDEBUG(D_HA, "%s retrieved update log, duration %lld, retries %d\n",
 438                dt->dd_lu_dev.ld_obd->obd_name, ktime_get_real_seconds() - start,
 439                retries);
 440
 441         spin_lock(&lod->lod_lock);
 442         if (!lrd->lrd_ltd)
 443                 lod->lod_child_got_update_log = 1;
 444         else
 445                 lrd->lrd_ltd->ltd_got_update_log = 1;
 446
 447         if (!lod->lod_child_got_update_log) {
 448                 spin_unlock(&lod->lod_lock);
 449                 GOTO(out, rc = 0);
 450         }
 451
 452         lod_foreach_mdt(lod, mdt) {
 453                 if (!mdt->ltd_got_update_log) {
 454                         spin_unlock(&lod->lod_lock);
 455                         GOTO(out, rc = 0);
 456                 }
 457         }
 458         lut->lut_tdtd->tdtd_replay_ready = 1;
 459         spin_unlock(&lod->lod_lock);
 460
 461         CDEBUG(D_HA, "%s got update logs from all MDTs.\n",
 462                lut->lut_obd->obd_name);
 463         wake_up(&lut->lut_obd->obd_next_transno_waitq);
 464         EXIT;
 465
 466 out:
 467         atomic_dec(&lut->lut_tdtd->tdtd_recovery_threads_count);
 468         wake_up(&lut->lut_tdtd->tdtd_recovery_threads_waitq);
 469         if (xchg(lrd->lrd_task, NULL) == NULL)
 470                 /* Someone is waiting for us to finish, need
 471                  * to synchronize cleanly.
 472                  */
 473                 wait_var_event(lrd, kthread_should_stop());
 474         lu_env_fini(env);
 475         OBD_FREE_PTR(lrd);
 476         return 0;
 477 }
 478
 479 /**
 480  * finish sub llog context
 481  *
 482  * Stop update recovery thread for the sub device, then cleanup the
 483  * correspondent llog ctxt.
 484  *
 485  * \param[in] env      execution environment
 486  * \param[in] lod      lod device to do update recovery
 487  * \param[in] thread   recovery thread on this sub device
 488  */
 489 void lod_sub_fini_llog(const struct lu_env *env,
 490                        struct dt_device *dt, struct task_struct **thread)
 491 {
 492         struct obd_device *obd;
 493         struct llog_ctxt *ctxt;
 494         struct task_struct *task = NULL;
 495
 496         ENTRY;
 497
 498         obd = dt->dd_lu_dev.ld_obd;
 499         CDEBUG(D_INFO, "%s: finish sub llog\n", obd->obd_name);
 500         /* Wait for recovery thread to complete */
 501         if (thread)
 502                 task = xchg(thread, NULL);
 503         if (task)
 504                 kthread_stop(task);
 505
 506         ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
 507         if (!ctxt)
 508                 RETURN_EXIT;
 509
 510         if (ctxt->loc_handle)
 511                 llog_cat_close(env, ctxt->loc_handle);
 512
 513         llog_cleanup(env, ctxt);
 514
 515         RETURN_EXIT;
 516 }
 517
 518 /**
 519  * Extract MDT target index from a device name.
 520  *
 521  * a helper function to extract index from the given device name
 522  * like "fsname-MDTxxxx-mdtlov"
 523  *
 524  * \param[in] lodname           device name
 525  * \param[out] mdt_index        extracted index
 526  *
 527  * \retval 0            on success
 528  * \retval -EINVAL      if the name is invalid
 529  */
 530 int lodname2mdt_index(char *lodname, u32 *mdt_index)
 531 {
 532         u32 index;
 533         const char *ptr, *tmp;
 534         int rc;
 535
 536         /* 1.8 configs don't have "-MDT0000" at the end */
 537         ptr = strstr(lodname, "-MDT");
 538         if (!ptr) {
 539                 *mdt_index = 0;
 540                 return 0;
 541         }
 542
 543         ptr = strrchr(lodname, '-');
 544         if (!ptr) {
 545                 rc = -EINVAL;
 546                 CERROR("invalid MDT index in '%s': rc = %d\n", lodname, rc);
 547                 return rc;
 548         }
 549
 550         if (strncmp(ptr, "-mdtlov", 7) != 0) {
 551                 rc = -EINVAL;
 552                 CERROR("invalid MDT index in '%s': rc = %d\n", lodname, rc);
 553                 return rc;
 554         }
 555
 556         if ((unsigned long)ptr - (unsigned long)lodname <= 8) {
 557                 rc = -EINVAL;
 558                 CERROR("invalid MDT index in '%s': rc = %d\n", lodname, rc);
 559                 return rc;
 560         }
 561
 562         if (strncmp(ptr - 8, "-MDT", 4) != 0) {
 563                 rc = -EINVAL;
 564                 CERROR("invalid MDT index in '%s': rc = %d\n", lodname, rc);
 565                 return rc;
 566         }
 567
 568         rc = target_name2index(ptr - 7, &index, &tmp);
 569         if (rc < 0 || rc & LDD_F_SV_ALL || *tmp != '-') {
 570                 rc = -EINVAL;
 571                 CERROR("invalid MDT index in '%s': rc = %d\n", lodname, rc);
 572                 return rc;
 573         }
 574         *mdt_index = index;
 575         return 0;
 576 }
 577
 578 /**
 579  * Init sub llog context
 580  *
 581  * Setup update llog ctxt for update recovery threads, then start the
 582  * recovery thread (lod_sub_recovery_thread) to read update llog from
 583  * the correspondent MDT to do update recovery.
 584  *
 585  * \param[in] env       execution environment
 586  * \param[in] lod       lod device to do update recovery
 587  * \param[in] dt        sub dt device for which the recovery thread is
 588  *
 589  * \retval              0 if initialization succeeds.
 590  * \retval              negative errno if initialization fails.
 591  */
 592 int lod_sub_init_llog(const struct lu_env *env, struct lod_device *lod,
 593                       struct dt_device *dt)
 594 {
 595         struct obd_device *obd;
 596         struct lod_recovery_data *lrd = NULL;
 597         DECLARE_COMPLETION_ONSTACK(started);
 598         struct task_struct **taskp;
 599         struct task_struct *task;
 600         struct lod_tgt_desc *subtgt = NULL;
 601         u32 index;
 602         u32 master_index;
 603         int rc;
 604
 605         ENTRY;
 606
 607         rc = lodname2mdt_index(lod2obd(lod)->obd_name, &master_index);
 608         if (rc != 0)
 609                 RETURN(rc);
 610
 611         OBD_ALLOC_PTR(lrd);
 612         if (!lrd)
 613                 RETURN(-ENOMEM);
 614
 615         if (lod->lod_child == dt) {
 616                 taskp = &lod->lod_child_recovery_task;
 617                 index = master_index;
 618         } else {
 619                 struct lu_tgt_desc *mdt;
 620
 621                 lod_foreach_mdt(lod, mdt) {
 622                         if (mdt->ltd_tgt == dt) {
 623                                 index = mdt->ltd_index;
 624                                 subtgt = mdt;
 625                                 break;
 626                         }
 627                 }
 628                 LASSERT(subtgt != NULL);
 629                 taskp = &subtgt->ltd_recovery_task;
 630         }
 631
 632         CDEBUG(D_INFO, "%s init sub log %s\n", lod2obd(lod)->obd_name,
 633                dt->dd_lu_dev.ld_obd->obd_name);
 634         lrd->lrd_lod = lod;
 635         lrd->lrd_ltd = subtgt;
 636         lrd->lrd_task = taskp;
 637         lrd->lrd_idx = index;
 638         lrd->lrd_started = &started;
 639
 640         obd = dt->dd_lu_dev.ld_obd;
 641         obd->obd_lvfs_ctxt.dt = dt;
 642         rc = llog_setup(env, obd, &obd->obd_olg, LLOG_UPDATELOG_ORIG_CTXT,
 643                         NULL, &llog_common_cat_ops);
 644         if (rc < 0) {
 645                 CERROR("%s: cannot setup updatelog llog: rc = %d\n",
 646                        obd->obd_name, rc);
 647                 GOTO(free_lrd, rc);
 648         }
 649
 650         rc = lu_env_init(&lrd->lrd_env, LCT_LOCAL | LCT_MD_THREAD);
 651         if (rc != 0) {
 652                 CERROR("%s: can't initialize env: rc = %d\n",
 653                        lod2obd(lod)->obd_name, rc);
 654                 GOTO(free_lrd, rc);
 655         }
 656
 657         /* Start the recovery thread */
 658         task = kthread_create(lod_sub_recovery_thread, lrd, "lod%04x_rec%04x",
 659                               master_index, index);
 660         if (IS_ERR(task)) {
 661                 rc = PTR_ERR(task);
 662                 CERROR("%s: cannot start recovery thread: rc = %d\n",
 663                        obd->obd_name, rc);
 664                 lu_env_fini(&lrd->lrd_env);
 665                 GOTO(out_llog, rc);
 666         }
 667         *taskp = task;
 668         wake_up_process(task);
 669         wait_for_completion(&started);
 670
 671         RETURN(0);
 672 out_llog:
 673         lod_sub_fini_llog(env, dt, taskp);
 674 free_lrd:
 675         OBD_FREE_PTR(lrd);
 676         RETURN(rc);
 677 }
 678
 679 /**
 680  * Stop sub recovery thread
 681  *
 682  * Stop sub recovery thread on all subs.
 683  *
 684  * \param[in] env       execution environment
 685  * \param[in] lod       lod device to do update recovery
 686  */
 687 static void lod_sub_stop_recovery_threads(const struct lu_env *env,
 688                                           struct lod_device *lod)
 689 {
 690         struct task_struct *task;
 691         struct lu_tgt_desc *mdt;
 692
 693         /*
 694          * Stop the update log commit cancel threads and finish master
 695          * llog ctxt
 696          */
 697         task = xchg(&lod->lod_child_recovery_task, NULL);
 698         if (task)
 699                 kthread_stop(task);
 700
 701         lod_getref(&lod->lod_mdt_descs);
 702         lod_foreach_mdt(lod, mdt) {
 703                 task = xchg(&mdt->ltd_recovery_task, NULL);
 704                 if (task)
 705                         kthread_stop(task);
 706         }
 707         lod_putref(lod, &lod->lod_mdt_descs);
 708 }
 709
 710 /**
 711  * finish all sub llog
 712  *
 713  * cleanup all of sub llog ctxt on the LOD.
 714  *
 715  * \param[in] env       execution environment
 716  * \param[in] lod       lod device to do update recovery
 717  */
 718 static void lod_sub_fini_all_llogs(const struct lu_env *env,
 719                                    struct lod_device *lod)
 720 {
 721         struct lu_tgt_desc *mdt;
 722
 723         /*
 724          * Stop the update log commit cancel threads and finish master
 725          * llog ctxt
 726          */
 727         lod_sub_fini_llog(env, lod->lod_child,
 728                           &lod->lod_child_recovery_task);
 729         lod_getref(&lod->lod_mdt_descs);
 730         lod_foreach_mdt(lod, mdt)
 731                 lod_sub_fini_llog(env, mdt->ltd_tgt,
 732                                   &mdt->ltd_recovery_task);
 733         lod_putref(lod, &lod->lod_mdt_descs);
 734 }
 735
 736 static char *lod_show_update_logs_retrievers(void *data, int *size, int *count)
 737 {
 738         struct lod_device *lod = (struct lod_device *)data;
 739         struct lu_target *lut = lod2lu_dev(lod)->ld_site->ls_tgt;
 740         struct lu_tgt_desc *mdt = NULL;
 741         char *buf;
 742         int len = 0;
 743         int rc;
 744         int i;
 745
 746         *count = atomic_read(&lut->lut_tdtd->tdtd_recovery_threads_count);
 747         if (*count == 0) {
 748                 *size = 0;
 749                 return NULL;
 750         }
 751
 752         *size = 5 * *count + 1;
 753         OBD_ALLOC(buf, *size);
 754         if (!buf)
 755                 return NULL;
 756
 757         *count = 0;
 758         memset(buf, 0, *size);
 759
 760         if (!lod->lod_child_got_update_log) {
 761                 rc = lodname2mdt_index(lod2obd(lod)->obd_name, &i);
 762                 LASSERTF(rc == 0, "Fail to parse target index: rc = %d\n", rc);
 763
 764                 rc = scnprintf(buf + len, *size - len, " %04x", i);
 765                 LASSERT(rc > 0);
 766
 767                 len += rc;
 768                 (*count)++;
 769         }
 770
 771         lod_foreach_mdt(lod, mdt) {
 772                 if (!mdt->ltd_got_update_log) {
 773                         rc = scnprintf(buf + len, *size - len, " %04x",
 774                                        mdt->ltd_index);
 775                         if (unlikely(rc <= 0))
 776                                 break;
 777
 778                         len += rc;
 779                         (*count)++;
 780                 }
 781         }
 782
 783         return buf;
 784 }
 785
 786 /**
 787  * Prepare distribute txn
 788  *
 789  * Prepare distribute txn structure for LOD
 790  *
 791  * \param[in] env       execution environment
 792  * \param[in] lod_device  LOD device
 793  *
 794  * \retval              0 if preparation succeeds.
 795  * \retval              negative errno if preparation fails.
 796  */
 797 static int lod_prepare_distribute_txn(const struct lu_env *env,
 798                                       struct lod_device *lod)
 799 {
 800         struct target_distribute_txn_data *tdtd;
 801         struct lu_target *lut;
 802         int rc;
 803
 804         ENTRY;
 805
 806         /* Init update recovery data */
 807         OBD_ALLOC_PTR(tdtd);
 808         if (!tdtd)
 809                 RETURN(-ENOMEM);
 810
 811         lut = lod2lu_dev(lod)->ld_site->ls_tgt;
 812         tdtd->tdtd_dt = &lod->lod_dt_dev;
 813         rc = distribute_txn_init(env, lut, tdtd,
 814                 lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id);
 815
 816         if (rc < 0) {
 817                 CERROR("%s: cannot init distribute txn: rc = %d\n",
 818                        lod2obd(lod)->obd_name, rc);
 819                 OBD_FREE_PTR(tdtd);
 820                 RETURN(rc);
 821         }
 822
 823         tdtd->tdtd_show_update_logs_retrievers =
 824                 lod_show_update_logs_retrievers;
 825         tdtd->tdtd_show_retrievers_cbdata = lod;
 826
 827         lut->lut_tdtd = tdtd;
 828
 829         RETURN(0);
 830 }
 831
 832 /**
 833  * Finish distribute txn
 834  *
 835  * Release the resource holding by distribute txn, i.e. stop distribute
 836  * txn thread.
 837  *
 838  * \param[in] env       execution environment
 839  * \param[in] lod       lod device
 840  */
 841 static void lod_fini_distribute_txn(const struct lu_env *env,
 842                                     struct lod_device *lod)
 843 {
 844         struct lu_target *lut;
 845
 846         lut = lod2lu_dev(lod)->ld_site->ls_tgt;
 847         target_recovery_fini(lut->lut_obd);
 848         if (!lut->lut_tdtd)
 849                 return;
 850
 851         distribute_txn_fini(env, lut->lut_tdtd);
 852
 853         OBD_FREE_PTR(lut->lut_tdtd);
 854         lut->lut_tdtd = NULL;
 855 }
 856
 857 /**
 858  * Implementation of lu_device_operations::ldo_process_config() for LOD
 859  *
 860  * The method is called by the configuration subsystem during setup,
 861  * cleanup and when the configuration changes. The method processes
 862  * few specific commands like adding/removing the targets, changing
 863  * the runtime parameters.
 864
 865  * \param[in] env               LU environment provided by the caller
 866  * \param[in] dev               lod device
 867  * \param[in] lcfg              configuration command to apply
 868  *
 869  * \retval 0                    on success
 870  * \retval negative             negated errno on error
 871  *
 872  * The examples are below.
 873  *
 874  * Add osc config log:
 875  * marker  20 (flags=0x01, v2.2.49.56) lustre-OST0001  'add osc'
 876  * add_uuid  nid=192.168.122.162@tcp(0x20000c0a87aa2)  0:  1:nidxxx
 877  * attach    0:lustre-OST0001-osc-MDT0001  1:osc  2:lustre-MDT0001-mdtlov_UUID
 878  * setup     0:lustre-OST0001-osc-MDT0001  1:lustre-OST0001_UUID  2:nid
 879  * lov_modify_tgts add 0:lustre-MDT0001-mdtlov  1:lustre-OST0001_UUID  2:1  3:1
 880  * marker  20 (flags=0x02, v2.2.49.56) lustre-OST0001  'add osc'
 881  *
 882  * Add mdc config log:
 883  * marker  10 (flags=0x01, v2.2.49.56) lustre-MDT0000  'add osp'
 884  * add_uuid  nid=192.168.122.162@tcp(0x20000c0a87aa2)  0:  1:nid
 885  * attach 0:lustre-MDT0000-osp-MDT0001  1:osp  2:lustre-MDT0001-mdtlov_UUID
 886  * setup     0:lustre-MDT0000-osp-MDT0001  1:lustre-MDT0000_UUID  2:nid
 887  * modify_mdc_tgts add 0:lustre-MDT0001  1:lustre-MDT0000_UUID  2:0  3:1
 888  * marker  10 (flags=0x02, v2.2.49.56) lustre-MDT0000_UUID  'add osp'
 889  */
 890 static int lod_process_config(const struct lu_env *env,
 891                               struct lu_device *dev,
 892                               struct lustre_cfg *lcfg)
 893 {
 894         struct lod_device *lod = lu2lod_dev(dev);
 895         struct lu_device *next = &lod->lod_child->dd_lu_dev;
 896         char *arg1;
 897         int rc = 0;
 898
 899         ENTRY;
 900
 901         switch (lcfg->lcfg_command) {
 902         case LCFG_LOV_DEL_OBD:
 903         case LCFG_LOV_ADD_INA:
 904         case LCFG_LOV_ADD_OBD:
 905         case LCFG_ADD_MDC: {
 906                 u32 index;
 907                 u32 mdt_index;
 908                 int gen;
 909                 /*
 910                  * lov_modify_tgts add  0:lov_mdsA  1:osp  2:0  3:1
 911                  * modify_mdc_tgts add  0:lustre-MDT0001
 912                  *                    1:lustre-MDT0001-mdc0002
 913                  *                    2:2  3:1
 914                  */
 915                 arg1 = lustre_cfg_string(lcfg, 1);
 916
 917                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
 918                         GOTO(out, rc = -EINVAL);
 919                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
 920                         GOTO(out, rc = -EINVAL);
 921
 922                 if (lcfg->lcfg_command == LCFG_LOV_ADD_OBD) {
 923                         u32 mdt_index;
 924
 925                         rc = lodname2mdt_index(lustre_cfg_string(lcfg, 0),
 926                                                &mdt_index);
 927                         if (rc != 0)
 928                                 GOTO(out, rc);
 929
 930                         rc = lod_add_device(env, lod, arg1, index, gen,
 931                                             mdt_index, LUSTRE_OSC_NAME, 1);
 932                 } else if (lcfg->lcfg_command == LCFG_ADD_MDC) {
 933                         mdt_index = index;
 934                         rc = lod_add_device(env, lod, arg1, index, gen,
 935                                             mdt_index, LUSTRE_MDC_NAME, 1);
 936                 } else if (lcfg->lcfg_command == LCFG_LOV_ADD_INA) {
 937                         /*FIXME: Add mdt_index for LCFG_LOV_ADD_INA*/
 938                         mdt_index = 0;
 939                         rc = lod_add_device(env, lod, arg1, index, gen,
 940                                             mdt_index, LUSTRE_OSC_NAME, 0);
 941                 } else {
 942                         rc = lod_del_device(env, lod, &lod->lod_ost_descs,
 943                                             arg1, index, gen);
 944                 }
 945
 946                 break;
 947         }
 948
 949         case LCFG_PARAM: {
 950                 struct obd_device *obd;
 951                 ssize_t count;
 952                 char *param;
 953
 954                 /*
 955                  * Check if it is activate/deactivate mdc
 956                  * lustre-MDTXXXX-osp-MDTXXXX.active=1
 957                  */
 958                 param = lustre_cfg_buf(lcfg, 1);
 959                 if (strstr(param, "osp") && strstr(param, ".active=")) {
 960                         struct lod_tgt_desc *sub_tgt = NULL;
 961                         struct lu_tgt_desc *mdt;
 962                         char *ptr;
 963                         char *tmp;
 964
 965                         ptr = strstr(param, ".");
 966                         *ptr = '\0';
 967                         obd = class_name2obd(param);
 968                         if (!obd) {
 969                                 CERROR("%s: can not find %s: rc = %d\n",
 970                                        lod2obd(lod)->obd_name, param, -EINVAL);
 971                                 *ptr = '.';
 972                                 GOTO(out, rc);
 973                         }
 974
 975                         lod_foreach_mdt(lod, mdt) {
 976                                 if (mdt->ltd_tgt->dd_lu_dev.ld_obd == obd) {
 977                                         sub_tgt = mdt;
 978                                         break;
 979                                 }
 980                         }
 981
 982                         if (!sub_tgt) {
 983                                 CERROR("%s: can not find %s: rc = %d\n",
 984                                        lod2obd(lod)->obd_name, param, -EINVAL);
 985                                 *ptr = '.';
 986                                 GOTO(out, rc);
 987                         }
 988
 989                         *ptr = '.';
 990                         tmp = strstr(param, "=");
 991                         tmp++;
 992                         if (*tmp == '1') {
 993                                 struct llog_ctxt *ctxt;
 994
 995                                 obd = sub_tgt->ltd_tgt->dd_lu_dev.ld_obd;
 996                                 ctxt = llog_get_context(obd,
 997                                                 LLOG_UPDATELOG_ORIG_CTXT);
 998                                 if (!ctxt) {
 999                                         rc = llog_setup(env, obd, &obd->obd_olg,
1000                                                        LLOG_UPDATELOG_ORIG_CTXT,
1001                                                     NULL, &llog_common_cat_ops);
1002                                         if (rc < 0)
1003                                                 GOTO(out, rc);
1004                                 } else {
1005                                         llog_ctxt_put(ctxt);
1006                                 }
1007                                 rc = lod_sub_prep_llog(env, lod,
1008                                                        sub_tgt->ltd_tgt,
1009                                                        sub_tgt->ltd_index);
1010                                 if (rc == 0)
1011                                         sub_tgt->ltd_active = 1;
1012                         } else {
1013                                 lod_sub_fini_llog(env, sub_tgt->ltd_tgt,
1014                                                   NULL);
1015                                 sub_tgt->ltd_active = 0;
1016                         }
1017                         GOTO(out, rc);
1018                 }
1019
1020
1021                 if (strstr(param, PARAM_LOD) != NULL)
1022                         count = class_modify_config(lcfg, PARAM_LOD,
1023                                                     &lod->lod_dt_dev.dd_kobj);
1024                 else
1025                         count = class_modify_config(lcfg, PARAM_LOV,
1026                                                     &lod->lod_dt_dev.dd_kobj);
1027                 rc = count > 0 ? 0 : count;
1028                 GOTO(out, rc);
1029         }
1030         case LCFG_PRE_CLEANUP: {
1031                 lod_sub_process_config(env, lod, &lod->lod_mdt_descs, lcfg);
1032                 lod_sub_process_config(env, lod, &lod->lod_ost_descs, lcfg);
1033                 OBD_FAIL_TIMEOUT(OBD_FAIL_TGT_RECOVERY_CONNECT, cfs_fail_val * 2);
1034                 next = &lod->lod_child->dd_lu_dev;
1035                 rc = next->ld_ops->ldo_process_config(env, next, lcfg);
1036                 if (rc != 0)
1037                         CDEBUG(D_HA, "%s: can't process %u: %d\n",
1038                                lod2obd(lod)->obd_name, lcfg->lcfg_command, rc);
1039
1040                 lod_sub_stop_recovery_threads(env, lod);
1041                 lod_fini_distribute_txn(env, lod);
1042                 lod_sub_fini_all_llogs(env, lod);
1043                 break;
1044         }
1045         case LCFG_CLEANUP: {
1046                 if (lod->lod_md_root) {
1047                         dt_object_put(env, &lod->lod_md_root->ldo_obj);
1048                         lod->lod_md_root = NULL;
1049                 }
1050
1051                 /*
1052                  * do cleanup on underlying storage only when
1053                  * all OSPs are cleaned up, as they use that OSD as well
1054                  */
1055                 lu_dev_del_linkage(dev->ld_site, dev);
1056                 lod_sub_process_config(env, lod, &lod->lod_mdt_descs, lcfg);
1057                 lod_sub_process_config(env, lod, &lod->lod_ost_descs, lcfg);
1058                 next = &lod->lod_child->dd_lu_dev;
1059                 rc = next->ld_ops->ldo_process_config(env, next, lcfg);
1060                 if (rc)
1061                         CERROR("%s: can't process %u: rc = %d\n",
1062                                lod2obd(lod)->obd_name, lcfg->lcfg_command, rc);
1063
1064                 rc = obd_disconnect(lod->lod_child_exp);
1065                 if (rc)
1066                         CERROR("error in disconnect from storage: rc = %d\n",
1067                                rc);
1068                 break;
1069         }
1070         default:
1071                 CERROR("%s: unknown command %u\n", lod2obd(lod)->obd_name,
1072                        lcfg->lcfg_command);
1073                 rc = -EINVAL;
1074                 break;
1075         }
1076
1077 out:
1078         RETURN(rc);
1079 }
1080
1081 /**
1082  * Implementation of lu_device_operations::ldo_recovery_complete() for LOD
1083  *
1084  * The method is called once the recovery is complete. This implementation
1085  * distributes the notification to all the known targets.
1086  *
1087  * see include/lu_object.h for the details
1088  */
1089 static int lod_recovery_complete(const struct lu_env *env,
1090                                  struct lu_device *dev)
1091 {
1092         struct lod_device *lod = lu2lod_dev(dev);
1093         struct lu_device *next = &lod->lod_child->dd_lu_dev;
1094         struct lod_tgt_desc *tgt;
1095         int rc;
1096
1097         ENTRY;
1098
1099         LASSERT(lod->lod_recovery_completed == 0);
1100         lod->lod_recovery_completed = 1;
1101
1102         rc = next->ld_ops->ldo_recovery_complete(env, next);
1103
1104         lod_getref(&lod->lod_ost_descs);
1105         if (lod->lod_ost_descs.ltd_tgts_size > 0) {
1106                 lod_foreach_ost(lod, tgt) {
1107                         LASSERT(tgt && tgt->ltd_tgt);
1108                         next = &tgt->ltd_tgt->dd_lu_dev;
1109                         rc = next->ld_ops->ldo_recovery_complete(env, next);
1110                         if (rc)
1111                                 CERROR("%s: can't complete recovery on #%d: rc = %d\n",
1112                                        lod2obd(lod)->obd_name, tgt->ltd_index,
1113                                        rc);
1114                 }
1115         }
1116         lod_putref(lod, &lod->lod_ost_descs);
1117         RETURN(rc);
1118 }
1119
1120 /**
1121  * Init update logs on all sub device
1122  *
1123  * LOD initialize update logs on all of sub devices. Because the initialization
1124  * process might need FLD lookup, see llog_osd_open()->dt_locate()->...->
1125  * lod_object_init(), this API has to be called after LOD is initialized.
1126  * \param[in] env       execution environment
1127  * \param[in] lod       lod device
1128  *
1129  * \retval              0 if update log is initialized successfully.
1130  * \retval              negative errno if initialization fails.
1131  */
1132 static int lod_sub_init_llogs(const struct lu_env *env, struct lod_device *lod)
1133 {
1134         struct lu_tgt_desc *mdt;
1135         int rc;
1136
1137         ENTRY;
1138
1139         /*
1140          * llog must be setup after LOD is initialized, because llog
1141          * initialization include FLD lookup
1142          */
1143         LASSERT(lod->lod_initialized);
1144
1145         /* Init the llog in its own stack */
1146         rc = lod_sub_init_llog(env, lod, lod->lod_child);
1147         if (rc < 0)
1148                 RETURN(rc);
1149
1150         lod_foreach_mdt(lod, mdt) {
1151                 rc = lod_sub_init_llog(env, lod, mdt->ltd_tgt);
1152                 if (rc != 0)
1153                         break;
1154         }
1155
1156         RETURN(rc);
1157 }
1158
1159 /**
1160  * Implementation of lu_device_operations::ldo_prepare() for LOD
1161  *
1162  * see include/lu_object.h for the details.
1163  */
1164 static int lod_prepare(const struct lu_env *env, struct lu_device *pdev,
1165                        struct lu_device *cdev)
1166 {
1167         struct lod_device *lod = lu2lod_dev(cdev);
1168         struct lu_device *next = &lod->lod_child->dd_lu_dev;
1169         struct lu_fid *fid = &lod_env_info(env)->lti_fid;
1170         int rc;
1171         struct dt_object *root;
1172         struct dt_object *dto;
1173         u32 index;
1174
1175         ENTRY;
1176
1177         rc = next->ld_ops->ldo_prepare(env, pdev, next);
1178         if (rc != 0) {
1179                 CERROR("%s: prepare bottom error: rc = %d\n",
1180                        lod2obd(lod)->obd_name, rc);
1181                 RETURN(rc);
1182         }
1183
1184         lod->lod_initialized = 1;
1185
1186         rc = dt_root_get(env, lod->lod_child, fid);
1187         if (rc < 0)
1188                 RETURN(rc);
1189
1190         root = dt_locate(env, lod->lod_child, fid);
1191         if (IS_ERR(root))
1192                 RETURN(PTR_ERR(root));
1193
1194         /* Create update log object */
1195         index = lu_site2seq(lod2lu_dev(lod)->ld_site)->ss_node_id;
1196         lu_update_log_fid(fid, index);
1197
1198         dto = local_file_find_or_create_with_fid(env, lod->lod_child,
1199                                                  fid, root,
1200                                                  lod_update_log_name,
1201                                                  S_IFREG | 0644);
1202         if (IS_ERR(dto))
1203                 GOTO(out_put, rc = PTR_ERR(dto));
1204
1205         dt_object_put(env, dto);
1206
1207         /* Create update log dir */
1208         lu_update_log_dir_fid(fid, index);
1209         dto = local_file_find_or_create_with_fid(env, lod->lod_child,
1210                                                  fid, root,
1211                                                  lod_update_log_dir_name,
1212                                                  S_IFDIR | 0644);
1213         if (IS_ERR(dto))
1214                 GOTO(out_put, rc = PTR_ERR(dto));
1215
1216         dt_object_put(env, dto);
1217
1218         rc = lod_prepare_distribute_txn(env, lod);
1219         if (rc != 0)
1220                 GOTO(out_put, rc);
1221
1222         rc = lod_sub_init_llogs(env, lod);
1223         if (rc != 0)
1224                 GOTO(out_put, rc);
1225
1226 out_put:
1227         dt_object_put(env, root);
1228
1229         RETURN(rc);
1230 }
1231
1232 const struct lu_device_operations lod_lu_ops = {
1233         .ldo_object_alloc       = lod_object_alloc,
1234         .ldo_process_config     = lod_process_config,
1235         .ldo_recovery_complete  = lod_recovery_complete,
1236         .ldo_prepare            = lod_prepare,
1237 };
1238
1239 /**
1240  * Implementation of dt_device_operations::dt_root_get() for LOD
1241  *
1242  * see include/dt_object.h for the details.
1243  */
1244 static int lod_root_get(const struct lu_env *env,
1245                         struct dt_device *dev, struct lu_fid *f)
1246 {
1247         return dt_root_get(env, dt2lod_dev(dev)->lod_child, f);
1248 }
1249
1250 static void lod_statfs_sum(struct obd_statfs *sfs,
1251                              struct obd_statfs *ost_sfs, int *bs)
1252 {
1253         while (ost_sfs->os_bsize < *bs) {
1254                 *bs >>= 1;
1255                 sfs->os_bsize >>= 1;
1256                 sfs->os_bavail <<= 1;
1257                 sfs->os_blocks <<= 1;
1258                 sfs->os_bfree <<= 1;
1259                 sfs->os_granted <<= 1;
1260         }
1261         while (ost_sfs->os_bsize > *bs) {
1262                 ost_sfs->os_bsize >>= 1;
1263                 ost_sfs->os_bavail <<= 1;
1264                 ost_sfs->os_blocks <<= 1;
1265                 ost_sfs->os_bfree <<= 1;
1266                 ost_sfs->os_granted <<= 1;
1267         }
1268         sfs->os_bavail += ost_sfs->os_bavail;
1269         sfs->os_blocks += ost_sfs->os_blocks;
1270         sfs->os_bfree += ost_sfs->os_bfree;
1271         sfs->os_granted += ost_sfs->os_granted;
1272 }
1273
1274 /**
1275  * Implementation of dt_device_operations::dt_statfs() for LOD
1276  *
1277  * see include/dt_object.h for the details.
1278  */
1279 static int lod_statfs(const struct lu_env *env, struct dt_device *dev,
1280                       struct obd_statfs *sfs, struct obd_statfs_info *info)
1281 {
1282         struct lod_device *lod = dt2lod_dev(dev);
1283         struct lu_tgt_desc *tgt;
1284         struct obd_statfs ost_sfs;
1285         u64 ost_files = 0;
1286         u64 ost_ffree = 0;
1287         int rc, bs;
1288
1289         rc = dt_statfs(env, dt2lod_dev(dev)->lod_child, sfs);
1290         if (rc)
1291                 GOTO(out, rc);
1292
1293         bs = sfs->os_bsize;
1294
1295         sfs->os_bavail = 0;
1296         sfs->os_blocks = 0;
1297         sfs->os_bfree = 0;
1298         sfs->os_granted = 0;
1299
1300         lod_getref(&lod->lod_mdt_descs);
1301         lod_foreach_mdt(lod, tgt) {
1302                 rc = dt_statfs(env, tgt->ltd_tgt, &ost_sfs);
1303                 /* ignore errors */
1304                 if (rc)
1305                         continue;
1306                 sfs->os_files += ost_sfs.os_files;
1307                 sfs->os_ffree += ost_sfs.os_ffree;
1308                 lod_statfs_sum(sfs, &ost_sfs, &bs);
1309         }
1310         lod_putref(lod, &lod->lod_mdt_descs);
1311
1312         /*
1313          * at some point we can check whether DoM is enabled and
1314          * decide how to account MDT space. for simplicity let's
1315          * just fallback to pre-DoM policy if any OST is alive
1316          */
1317         lod_getref(&lod->lod_ost_descs);
1318         lod_foreach_ost(lod, tgt) {
1319                 rc = dt_statfs(env, tgt->ltd_tgt, &ost_sfs);
1320                 /* ignore errors */
1321                 if (rc || ost_sfs.os_bsize == 0)
1322                         continue;
1323                 if (!ost_files) {
1324                         /*
1325                          * if only MDTs with DoM then report only MDT blocks,
1326                          * otherwise show only OST blocks, and DoM is "free"
1327                          */
1328                         sfs->os_bavail = 0;
1329                         sfs->os_blocks = 0;
1330                         sfs->os_bfree = 0;
1331                         sfs->os_granted = 0;
1332                 }
1333                 ost_files += ost_sfs.os_files;
1334                 ost_ffree += ost_sfs.os_ffree;
1335                 ost_sfs.os_bavail += ost_sfs.os_granted;
1336                 lod_statfs_sum(sfs, &ost_sfs, &bs);
1337                 LASSERTF(bs == ost_sfs.os_bsize, "%d != %d\n",
1338                         (int)sfs->os_bsize, (int)ost_sfs.os_bsize);
1339         }
1340         lod_putref(lod, &lod->lod_ost_descs);
1341         sfs->os_state |= OS_STATE_SUM;
1342
1343         /* If we have _some_ OSTs, but don't have as many free objects on the
1344          * OSTs as inodes on the MDTs, reduce the reported number of inodes
1345          * to compensate, so that the "inodes in use" number is correct.
1346          * This should be kept in sync with ll_statfs_internal().
1347          */
1348         if (ost_files && ost_ffree < sfs->os_ffree) {
1349                 sfs->os_files = (sfs->os_files - sfs->os_ffree) + ost_ffree;
1350                 sfs->os_ffree = ost_ffree;
1351         }
1352
1353         /* a single successful statfs should be enough */
1354         rc = 0;
1355
1356 out:
1357         RETURN(rc);
1358 }
1359
1360 /**
1361  * Implementation of dt_device_operations::dt_trans_create() for LOD
1362  *
1363  * Creates a transaction using local (to this node) OSD.
1364  *
1365  * see include/dt_object.h for the details.
1366  */
1367 static struct thandle *lod_trans_create(const struct lu_env *env,
1368                                         struct dt_device *dt)
1369 {
1370         struct thandle *th;
1371
1372         th = top_trans_create(env, dt2lod_dev(dt)->lod_child);
1373         if (IS_ERR(th))
1374                 return th;
1375
1376         th->th_dev = dt;
1377
1378         return th;
1379 }
1380
1381 /**
1382  * Implementation of dt_device_operations::dt_trans_start() for LOD
1383  *
1384  * Starts the set of local transactions using the targets involved
1385  * in declare phase. Initial support for the distributed transactions.
1386  *
1387  * see include/dt_object.h for the details.
1388  */
1389 static int lod_trans_start(const struct lu_env *env, struct dt_device *dt,
1390                            struct thandle *th)
1391 {
1392         return top_trans_start(env, dt2lod_dev(dt)->lod_child, th);
1393 }
1394
1395 static int lod_trans_cb_add(struct thandle *th,
1396                             struct dt_txn_commit_cb *dcb)
1397 {
1398         struct top_thandle      *top_th = container_of(th, struct top_thandle,
1399                                                        tt_super);
1400         return dt_trans_cb_add(top_th->tt_master_sub_thandle, dcb);
1401 }
1402
1403 /**
1404  * add noop update to the update records
1405  *
1406  * Add noop updates to the update records, which is only used in
1407  * test right now.
1408  *
1409  * \param[in] env       execution environment
1410  * \param[in] dt        dt device of lod
1411  * \param[in] th        thandle
1412  * \param[in] count     the count of update records to be added.
1413  *
1414  * \retval              0 if adding succeeds.
1415  * \retval              negative errno if adding fails.
1416  */
1417 static int lod_add_noop_records(const struct lu_env *env,
1418                                 struct dt_device *dt, struct thandle *th,
1419                                 int count)
1420 {
1421         struct top_thandle *top_th;
1422         struct lu_fid *fid = &lod_env_info(env)->lti_fid;
1423         int i;
1424         int rc = 0;
1425
1426         top_th = container_of(th, struct top_thandle, tt_super);
1427         if (!top_th->tt_multiple_thandle)
1428                 return 0;
1429
1430         fid_zero(fid);
1431         for (i = 0; i < count; i++) {
1432                 rc = update_record_pack(noop, th, fid);
1433                 if (rc < 0)
1434                         return rc;
1435         }
1436         return rc;
1437 }
1438
1439 /**
1440  * Implementation of dt_device_operations::dt_trans_stop() for LOD
1441  *
1442  * Stops the set of local transactions using the targets involved
1443  * in declare phase. Initial support for the distributed transactions.
1444  *
1445  * see include/dt_object.h for the details.
1446  */
1447 static int lod_trans_stop(const struct lu_env *env, struct dt_device *dt,
1448                           struct thandle *th)
1449 {
1450         if (OBD_FAIL_CHECK(OBD_FAIL_SPLIT_UPDATE_REC)) {
1451                 int rc;
1452
1453                 rc = lod_add_noop_records(env, dt, th, 5000);
1454                 if (rc < 0)
1455                         RETURN(rc);
1456         }
1457         return top_trans_stop(env, dt2lod_dev(dt)->lod_child, th);
1458 }
1459
1460 /**
1461  * Implementation of dt_device_operations::dt_conf_get() for LOD
1462  *
1463  * Currently returns the configuration provided by the local OSD.
1464  *
1465  * see include/dt_object.h for the details.
1466  */
1467 static void lod_conf_get(const struct lu_env *env,
1468                          const struct dt_device *dev,
1469                          struct dt_device_param *param)
1470 {
1471         dt_conf_get(env, dt2lod_dev((struct dt_device *)dev)->lod_child, param);
1472 }
1473
1474 /**
1475  * Implementation of dt_device_operations::dt_sync() for LOD
1476  *
1477  * Syncs all known OST targets. Very very expensive and used
1478  * rarely by LFSCK now. Should not be used in general.
1479  *
1480  * see include/dt_object.h for the details.
1481  */
1482 static int lod_sync(const struct lu_env *env, struct dt_device *dev)
1483 {
1484         struct lod_device *lod = dt2lod_dev(dev);
1485         struct lu_tgt_desc *tgt;
1486         int rc = 0;
1487
1488         ENTRY;
1489
1490         lod_getref(&lod->lod_ost_descs);
1491         lod_foreach_ost(lod, tgt) {
1492                 if (!tgt->ltd_active)
1493                         continue;
1494                 rc = dt_sync(env, tgt->ltd_tgt);
1495                 if (rc) {
1496                         if (rc != -ENOTCONN) {
1497                                 CERROR("%s: can't sync ost %u: rc = %d\n",
1498                                        lod2obd(lod)->obd_name, tgt->ltd_index,
1499                                        rc);
1500                                 break;
1501                         }
1502                         rc = 0;
1503                 }
1504         }
1505         lod_putref(lod, &lod->lod_ost_descs);
1506
1507         if (rc)
1508                 RETURN(rc);
1509
1510         lod_getref(&lod->lod_mdt_descs);
1511         lod_foreach_mdt(lod, tgt) {
1512                 if (!tgt->ltd_active)
1513                         continue;
1514                 rc = dt_sync(env, tgt->ltd_tgt);
1515                 if (rc) {
1516                         if (rc != -ENOTCONN) {
1517                                 CERROR("%s: can't sync mdt %u: rc = %d\n",
1518                                        lod2obd(lod)->obd_name, tgt->ltd_index,
1519                                        rc);
1520                                 break;
1521                         }
1522                         rc = 0;
1523                 }
1524         }
1525         lod_putref(lod, &lod->lod_mdt_descs);
1526
1527         if (rc == 0)
1528                 rc = dt_sync(env, lod->lod_child);
1529
1530         RETURN(rc);
1531 }
1532
1533 /**
1534  * Implementation of dt_device_operations::dt_ro() for LOD
1535  *
1536  * Turns local OSD read-only, used for the testing only.
1537  *
1538  * see include/dt_object.h for the details.
1539  */
1540 static int lod_ro(const struct lu_env *env, struct dt_device *dev)
1541 {
1542         return dt_ro(env, dt2lod_dev(dev)->lod_child);
1543 }
1544
1545 /**
1546  * Implementation of dt_device_operations::dt_commit_async() for LOD
1547  *
1548  * Asks local OSD to commit sooner.
1549  *
1550  * see include/dt_object.h for the details.
1551  */
1552 static int lod_commit_async(const struct lu_env *env, struct dt_device *dev)
1553 {
1554         return dt_commit_async(env, dt2lod_dev(dev)->lod_child);
1555 }
1556
1557 static const struct dt_device_operations lod_dt_ops = {
1558         .dt_root_get         = lod_root_get,
1559         .dt_statfs           = lod_statfs,
1560         .dt_trans_create     = lod_trans_create,
1561         .dt_trans_start      = lod_trans_start,
1562         .dt_trans_stop       = lod_trans_stop,
1563         .dt_conf_get         = lod_conf_get,
1564         .dt_sync             = lod_sync,
1565         .dt_ro               = lod_ro,
1566         .dt_commit_async     = lod_commit_async,
1567         .dt_trans_cb_add     = lod_trans_cb_add,
1568 };
1569
1570 /**
1571  * Connect to a local OSD.
1572  *
1573  * Used to connect to the local OSD at mount. OSD name is taken from the
1574  * configuration command passed. This connection is used to identify LU
1575  * site and pin the OSD from early removal.
1576  *
1577  * \param[in] env               LU environment provided by the caller
1578  * \param[in] lod               lod device
1579  * \param[in] cfg               configuration command to apply
1580  *
1581  * \retval 0                    on success
1582  * \retval negative             negated errno on error
1583  **/
1584 static int lod_connect_to_osd(const struct lu_env *env, struct lod_device *lod,
1585                               struct lustre_cfg *cfg)
1586 {
1587         struct obd_connect_data *data = NULL;
1588         struct obd_device *obd;
1589         char *nextdev = NULL, *p, *s;
1590         int rc, len = 0;
1591
1592         ENTRY;
1593
1594         LASSERT(cfg);
1595         LASSERT(lod->lod_child_exp == NULL);
1596
1597         /*
1598          * compatibility hack: we still use old config logs
1599          * which specify LOV, but we need to learn underlying
1600          * OSD device, which is supposed to be:
1601          *  <fsname>-MDTxxxx-osd
1602          *
1603          * 2.x MGS generates lines like the following:
1604          *   #03 (176)lov_setup 0:lustre-MDT0000-mdtlov  1:(struct lov_desc)
1605          * 1.8 MGS generates lines like the following:
1606          *   #03 (168)lov_setup 0:lustre-mdtlov  1:(struct lov_desc)
1607          *
1608          * we use "-MDT" to differentiate 2.x from 1.8
1609          */
1610         p = lustre_cfg_string(cfg, 0);
1611         if (p && strstr(p, "-mdtlov")) {
1612                 len = strlen(p) + 6;
1613                 OBD_ALLOC(nextdev, len);
1614                 if (!nextdev)
1615                         GOTO(out, rc = -ENOMEM);
1616
1617                 strcpy(nextdev, p);
1618                 s = strstr(nextdev, "-mdtlov");
1619                 if (unlikely(!s)) {
1620                         CERROR("%s: unable to parse device name: rc = %d\n",
1621                                lustre_cfg_string(cfg, 0), -EINVAL);
1622                         GOTO(out, rc = -EINVAL);
1623                 }
1624
1625                 if (strstr(nextdev, "-MDT")) {
1626                         /* 2.x config */
1627                         strcpy(s, "-osd");
1628                 } else {
1629                         /* 1.8 config */
1630                         strcpy(s, "-MDT0000-osd");
1631                 }
1632         } else {
1633                 CERROR("%s: unable to parse device name: rc = %d\n",
1634                        lustre_cfg_string(cfg, 0), -EINVAL);
1635                 GOTO(out, rc = -EINVAL);
1636         }
1637
1638         OBD_ALLOC_PTR(data);
1639         if (!data)
1640                 GOTO(out, rc = -ENOMEM);
1641
1642         obd = class_name2obd(nextdev);
1643         if (!obd) {
1644                 CERROR("%s: can not locate next device: rc = %d\n",
1645                        nextdev, -ENOTCONN);
1646                 GOTO(out, rc = -ENOTCONN);
1647         }
1648
1649         data->ocd_connect_flags = OBD_CONNECT_VERSION;
1650         data->ocd_version = LUSTRE_VERSION_CODE;
1651
1652         rc = obd_connect(env, &lod->lod_child_exp, obd, &obd->obd_uuid,
1653                          data, NULL);
1654         if (rc) {
1655                 CERROR("%s: cannot connect to next dev: rc = %d\n",
1656                        nextdev, rc);
1657                 GOTO(out, rc);
1658         }
1659
1660         lod->lod_dt_dev.dd_lu_dev.ld_site =
1661                 lod->lod_child_exp->exp_obd->obd_lu_dev->ld_site;
1662         LASSERT(lod->lod_dt_dev.dd_lu_dev.ld_site);
1663         lod->lod_child = lu2dt_dev(lod->lod_child_exp->exp_obd->obd_lu_dev);
1664
1665 out:
1666         if (data)
1667                 OBD_FREE_PTR(data);
1668         if (nextdev)
1669                 OBD_FREE(nextdev, len);
1670         RETURN(rc);
1671 }
1672
1673 /**
1674  * Initialize LOD device at setup.
1675  *
1676  * Initializes the given LOD device using the original configuration command.
1677  * The function initiates a connection to the local OSD and initializes few
1678  * internal structures like pools, target tables, etc.
1679  *
1680  * \param[in] env               LU environment provided by the caller
1681  * \param[in] lod               lod device
1682  * \param[in] ldt               not used
1683  * \param[in] cfg               configuration command
1684  *
1685  * \retval 0                    on success
1686  * \retval negative             negated errno on error
1687  **/
1688 static int lod_init0(const struct lu_env *env, struct lod_device *lod,
1689                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
1690 {
1691         struct dt_device_param ddp;
1692         struct obd_device *obd;
1693         int rc;
1694
1695         ENTRY;
1696
1697         obd = class_name2obd(lustre_cfg_string(cfg, 0));
1698         if (!obd) {
1699                 rc = -ENODEV;
1700                 CERROR("Cannot find obd with name '%s': rc = %d\n",
1701                        lustre_cfg_string(cfg, 0), rc);
1702                 RETURN(rc);
1703         }
1704
1705         obd->obd_lu_dev = &lod->lod_dt_dev.dd_lu_dev;
1706         lod->lod_dt_dev.dd_lu_dev.ld_obd = obd;
1707         lod->lod_dt_dev.dd_lu_dev.ld_ops = &lod_lu_ops;
1708         lod->lod_dt_dev.dd_ops = &lod_dt_ops;
1709
1710         rc = lod_connect_to_osd(env, lod, cfg);
1711         if (rc)
1712                 RETURN(rc);
1713
1714         dt_conf_get(env, &lod->lod_dt_dev, &ddp);
1715         lod->lod_osd_max_easize = ddp.ddp_max_ea_size;
1716         lod->lod_dom_max_stripesize = (1ULL << 20); /* 1Mb as default value */
1717
1718         /* setup obd to be used with old lov code */
1719         rc = lod_pools_init(lod, cfg);
1720         if (rc)
1721                 GOTO(out_disconnect, rc);
1722
1723         rc = lod_procfs_init(lod);
1724         if (rc)
1725                 GOTO(out_pools, rc);
1726
1727         spin_lock_init(&lod->lod_lock);
1728         spin_lock_init(&lod->lod_connects_lock);
1729         lu_tgt_descs_init(&lod->lod_mdt_descs, true);
1730         lu_tgt_descs_init(&lod->lod_ost_descs, false);
1731
1732         RETURN(0);
1733
1734 out_pools:
1735         lod_pools_fini(lod);
1736 out_disconnect:
1737         obd_disconnect(lod->lod_child_exp);
1738         RETURN(rc);
1739 }
1740
1741 /**
1742  * Implementation of lu_device_type_operations::ldto_device_free() for LOD
1743  *
1744  * Releases the memory allocated for LOD device.
1745  *
1746  * see include/lu_object.h for the details.
1747  */
1748 static struct lu_device *lod_device_free(const struct lu_env *env,
1749                                          struct lu_device *lu)
1750 {
1751         struct lod_device *lod = lu2lod_dev(lu);
1752         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
1753
1754         ENTRY;
1755
1756         if (atomic_read(&lu->ld_ref) > 0 &&
1757             !cfs_hash_is_empty(lu->ld_site->ls_obj_hash)) {
1758                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1759                 lu_site_print(env, lu->ld_site, &msgdata, lu_cdebug_printer);
1760         }
1761         LASSERTF(atomic_read(&lu->ld_ref) == 0, "lu is %p\n", lu);
1762         dt_device_fini(&lod->lod_dt_dev);
1763         OBD_FREE_PTR(lod);
1764         RETURN(next);
1765 }
1766
1767 /**
1768  * Implementation of lu_device_type_operations::ldto_device_alloc() for LOD
1769  *
1770  * Allocates LOD device and calls the helpers to initialize it.
1771  *
1772  * see include/lu_object.h for the details.
1773  */
1774 static struct lu_device *lod_device_alloc(const struct lu_env *env,
1775                                           struct lu_device_type *type,
1776                                           struct lustre_cfg *lcfg)
1777 {
1778         struct lod_device *lod;
1779         struct lu_device *lu_dev;
1780
1781         OBD_ALLOC_PTR(lod);
1782         if (!lod) {
1783                 lu_dev = ERR_PTR(-ENOMEM);
1784         } else {
1785                 int rc;
1786
1787                 lu_dev = lod2lu_dev(lod);
1788                 dt_device_init(&lod->lod_dt_dev, type);
1789                 rc = lod_init0(env, lod, type, lcfg);
1790                 if (rc != 0) {
1791                         lod_device_free(env, lu_dev);
1792                         lu_dev = ERR_PTR(rc);
1793                 }
1794         }
1795
1796         return lu_dev;
1797 }
1798
1799 static void lod_avoid_guide_fini(struct lod_avoid_guide *lag)
1800 {
1801         if (lag->lag_oss_avoid_array)
1802                 OBD_FREE(lag->lag_oss_avoid_array,
1803                          sizeof(u32) * lag->lag_oaa_size);
1804         if (lag->lag_ost_avoid_bitmap)
1805                 CFS_FREE_BITMAP(lag->lag_ost_avoid_bitmap);
1806 }
1807
1808 /**
1809  * Implementation of lu_device_type_operations::ldto_device_fini() for LOD
1810  *
1811  * Releases the internal resources used by LOD device.
1812  *
1813  * see include/lu_object.h for the details.
1814  */
1815 static struct lu_device *lod_device_fini(const struct lu_env *env,
1816                                          struct lu_device *d)
1817 {
1818         struct lod_device *lod = lu2lod_dev(d);
1819         int rc;
1820
1821         ENTRY;
1822
1823         lod_pools_fini(lod);
1824
1825         lod_procfs_fini(lod);
1826
1827         rc = lod_fini_tgt(env, lod, &lod->lod_ost_descs);
1828         if (rc)
1829                 CERROR("%s: can not fini ost descriptors: rc =  %d\n",
1830                         lod2obd(lod)->obd_name, rc);
1831
1832         rc = lod_fini_tgt(env, lod, &lod->lod_mdt_descs);
1833         if (rc)
1834                 CERROR("%s: can not fini mdt descriptors: rc =  %d\n",
1835                         lod2obd(lod)->obd_name, rc);
1836
1837         RETURN(NULL);
1838 }
1839
1840 /**
1841  * Implementation of obd_ops::o_connect() for LOD
1842  *
1843  * Used to track all the users of this specific LOD device,
1844  * so the device stays up until the last user disconnected.
1845  *
1846  * \param[in] env               LU environment provided by the caller
1847  * \param[out] exp              export the caller will be using to access LOD
1848  * \param[in] obd               OBD device representing LOD device
1849  * \param[in] cluuid            unique identifier of the caller
1850  * \param[in] data              not used
1851  * \param[in] localdata         not used
1852  *
1853  * \retval 0                    on success
1854  * \retval negative             negated errno on error
1855  **/
1856 static int lod_obd_connect(const struct lu_env *env, struct obd_export **exp,
1857                            struct obd_device *obd, struct obd_uuid *cluuid,
1858                            struct obd_connect_data *data, void *localdata)
1859 {
1860         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
1861         struct lustre_handle conn;
1862         int rc;
1863
1864         ENTRY;
1865
1866         CDEBUG(D_CONFIG, "connect #%d\n", lod->lod_connects);
1867
1868         rc = class_connect(&conn, obd, cluuid);
1869         if (rc)
1870                 RETURN(rc);
1871
1872         *exp = class_conn2export(&conn);
1873
1874         spin_lock(&lod->lod_connects_lock);
1875         lod->lod_connects++;
1876         /* at the moment we expect the only user */
1877         LASSERT(lod->lod_connects == 1);
1878         spin_unlock(&lod->lod_connects_lock);
1879
1880         RETURN(0);
1881 }
1882
1883 /**
1884  *
1885  * Implementation of obd_ops::o_disconnect() for LOD
1886  *
1887  * When the caller doesn't need to use this LOD instance, it calls
1888  * obd_disconnect() and LOD releases corresponding export/reference count.
1889  * Once all the users gone, LOD device is released.
1890  *
1891  * \param[in] exp               export provided to the caller in obd_connect()
1892  *
1893  * \retval 0                    on success
1894  * \retval negative             negated errno on error
1895  **/
1896 static int lod_obd_disconnect(struct obd_export *exp)
1897 {
1898         struct obd_device *obd = exp->exp_obd;
1899         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
1900         int rc, release = 0;
1901
1902         ENTRY;
1903
1904         /* Only disconnect the underlying layers on the final disconnect. */
1905         spin_lock(&lod->lod_connects_lock);
1906         lod->lod_connects--;
1907         if (lod->lod_connects != 0) {
1908                 /* why should there be more than 1 connect? */
1909                 spin_unlock(&lod->lod_connects_lock);
1910                 CERROR("%s: disconnect #%d\n", exp->exp_obd->obd_name,
1911                        lod->lod_connects);
1912                 goto out;
1913         }
1914         spin_unlock(&lod->lod_connects_lock);
1915
1916         /* the last user of lod has gone, let's release the device */
1917         release = 1;
1918
1919 out:
1920         rc = class_disconnect(exp); /* bz 9811 */
1921
1922         if (rc == 0 && release)
1923                 class_manual_cleanup(obd);
1924         RETURN(rc);
1925 }
1926
1927 LU_KEY_INIT(lod, struct lod_thread_info);
1928
1929 static void lod_key_fini(const struct lu_context *ctx,
1930                 struct lu_context_key *key, void *data)
1931 {
1932         struct lod_thread_info *info = data;
1933         struct lod_layout_component *lds =
1934                                 info->lti_def_striping.lds_def_comp_entries;
1935
1936         /*
1937          * allocated in lod_get_lov_ea
1938          * XXX: this is overload, a tread may have such store but used only
1939          * once. Probably better would be pool of such stores per LOD.
1940          */
1941         if (info->lti_ea_store) {
1942                 OBD_FREE_LARGE(info->lti_ea_store, info->lti_ea_store_size);
1943                 info->lti_ea_store = NULL;
1944                 info->lti_ea_store_size = 0;
1945         }
1946         lu_buf_free(&info->lti_linkea_buf);
1947
1948         if (lds)
1949                 lod_free_def_comp_entries(&info->lti_def_striping);
1950
1951         if (info->lti_comp_size > 0)
1952                 OBD_FREE(info->lti_comp_idx,
1953                          info->lti_comp_size * sizeof(u32));
1954
1955         lod_avoid_guide_fini(&info->lti_avoid);
1956
1957         OBD_FREE_PTR(info);
1958 }
1959
1960 /* context key: lod_thread_key */
1961 LU_CONTEXT_KEY_DEFINE(lod, LCT_MD_THREAD);
1962
1963 LU_TYPE_INIT_FINI(lod, &lod_thread_key);
1964
1965 static struct lu_device_type_operations lod_device_type_ops = {
1966         .ldto_init           = lod_type_init,
1967         .ldto_fini           = lod_type_fini,
1968
1969         .ldto_start          = lod_type_start,
1970         .ldto_stop           = lod_type_stop,
1971
1972         .ldto_device_alloc   = lod_device_alloc,
1973         .ldto_device_free    = lod_device_free,
1974
1975         .ldto_device_fini    = lod_device_fini
1976 };
1977
1978 static struct lu_device_type lod_device_type = {
1979         .ldt_tags     = LU_DEVICE_DT,
1980         .ldt_name     = LUSTRE_LOD_NAME,
1981         .ldt_ops      = &lod_device_type_ops,
1982         .ldt_ctx_tags = LCT_MD_THREAD,
1983 };
1984
1985 /**
1986  * Implementation of obd_ops::o_get_info() for LOD
1987  *
1988  * Currently, there is only one supported key: KEY_OSP_CONNECTED , to provide
1989  * the caller binary status whether LOD has seen connection to any OST target.
1990  * It will also check if the MDT update log context being initialized (if
1991  * needed).
1992  *
1993  * \param[in] env               LU environment provided by the caller
1994  * \param[in] exp               export of the caller
1995  * \param[in] keylen            len of the key
1996  * \param[in] key               the key
1997  * \param[in] vallen            not used
1998  * \param[in] val               not used
1999  *
2000  * \retval                      0 if a connection was seen
2001  * \retval                      -EAGAIN if LOD isn't running yet or no
2002  *                              connection has been seen yet
2003  * \retval                      -EINVAL if not supported key is requested
2004  **/
2005 static int lod_obd_get_info(const struct lu_env *env, struct obd_export *exp,
2006                             u32 keylen, void *key, u32 *vallen, void *val)
2007 {
2008         int rc = -EINVAL;
2009
2010         if (KEY_IS(KEY_OSP_CONNECTED)) {
2011                 struct obd_device *obd = exp->exp_obd;
2012                 struct lod_device *d;
2013                 struct lod_tgt_desc *tgt;
2014                 int rc = 1;
2015
2016                 if (!obd->obd_set_up || obd->obd_stopping)
2017                         RETURN(-EAGAIN);
2018
2019                 d = lu2lod_dev(obd->obd_lu_dev);
2020                 lod_getref(&d->lod_ost_descs);
2021                 lod_foreach_ost(d, tgt) {
2022                         rc = obd_get_info(env, tgt->ltd_exp, keylen, key,
2023                                           vallen, val);
2024                         /* one healthy device is enough */
2025                         if (rc == 0)
2026                                 break;
2027                 }
2028                 lod_putref(d, &d->lod_ost_descs);
2029
2030                 lod_getref(&d->lod_mdt_descs);
2031                 lod_foreach_mdt(d, tgt) {
2032                         struct llog_ctxt *ctxt;
2033
2034                         if (!tgt->ltd_active)
2035                                 continue;
2036
2037                         ctxt = llog_get_context(tgt->ltd_tgt->dd_lu_dev.ld_obd,
2038                                                 LLOG_UPDATELOG_ORIG_CTXT);
2039                         if (!ctxt) {
2040                                 CDEBUG(D_INFO, "%s: %s is not ready.\n",
2041                                        obd->obd_name,
2042                                       tgt->ltd_tgt->dd_lu_dev.ld_obd->obd_name);
2043                                 rc = -EAGAIN;
2044                                 break;
2045                         }
2046                         if (!ctxt->loc_handle) {
2047                                 CDEBUG(D_INFO, "%s: %s is not ready.\n",
2048                                        obd->obd_name,
2049                                       tgt->ltd_tgt->dd_lu_dev.ld_obd->obd_name);
2050                                 rc = -EAGAIN;
2051                                 llog_ctxt_put(ctxt);
2052                                 break;
2053                         }
2054                         llog_ctxt_put(ctxt);
2055                 }
2056                 lod_putref(d, &d->lod_mdt_descs);
2057
2058                 RETURN(rc);
2059         }
2060
2061         RETURN(rc);
2062 }
2063
2064 static int lod_obd_set_info_async(const struct lu_env *env,
2065                                   struct obd_export *exp,
2066                                   u32 keylen, void *key,
2067                                   u32 vallen, void *val,
2068                                   struct ptlrpc_request_set *set)
2069 {
2070         struct obd_device *obd = class_exp2obd(exp);
2071         struct lod_device *d;
2072         struct lod_tgt_desc *tgt;
2073         int no_set = 0;
2074         int rc = 0, rc2;
2075
2076         ENTRY;
2077
2078         if (!set) {
2079                 no_set = 1;
2080                 set = ptlrpc_prep_set();
2081                 if (!set)
2082                         RETURN(-ENOMEM);
2083         }
2084
2085         d = lu2lod_dev(obd->obd_lu_dev);
2086         lod_getref(&d->lod_ost_descs);
2087         lod_foreach_ost(d, tgt) {
2088                 if (!tgt->ltd_active)
2089                         continue;
2090
2091                 rc2 = obd_set_info_async(env, tgt->ltd_exp, keylen, key,
2092                                          vallen, val, set);
2093                 if (rc2 != 0 && rc == 0)
2094                         rc = rc2;
2095         }
2096         lod_putref(d, &d->lod_ost_descs);
2097
2098         lod_getref(&d->lod_mdt_descs);
2099         lod_foreach_mdt(d, tgt) {
2100                 if (!tgt->ltd_active)
2101                         continue;
2102                 rc2 = obd_set_info_async(env, tgt->ltd_exp, keylen, key,
2103                                          vallen, val, set);
2104                 if (rc2 != 0 && rc == 0)
2105                         rc = rc2;
2106         }
2107         lod_putref(d, &d->lod_mdt_descs);
2108
2109
2110         if (no_set) {
2111                 rc2 = ptlrpc_set_wait(env, set);
2112                 if (rc2 == 0 && rc == 0)
2113                         rc = rc2;
2114                 ptlrpc_set_destroy(set);
2115         }
2116         RETURN(rc);
2117 }
2118
2119 static const struct obd_ops lod_obd_device_ops = {
2120         .o_owner        = THIS_MODULE,
2121         .o_connect      = lod_obd_connect,
2122         .o_disconnect   = lod_obd_disconnect,
2123         .o_get_info     = lod_obd_get_info,
2124         .o_set_info_async = lod_obd_set_info_async,
2125         .o_pool_new     = lod_pool_new,
2126         .o_pool_rem     = lod_pool_remove,
2127         .o_pool_add     = lod_pool_add,
2128         .o_pool_del     = lod_pool_del,
2129 };
2130
2131 static int __init lod_init(void)
2132 {
2133         struct obd_type *sym;
2134         int rc;
2135
2136         rc = lu_kmem_init(lod_caches);
2137         if (rc)
2138                 return rc;
2139
2140         rc = class_register_type(&lod_obd_device_ops, NULL, true, NULL,
2141                                  LUSTRE_LOD_NAME, &lod_device_type);
2142         if (rc) {
2143                 lu_kmem_fini(lod_caches);
2144                 return rc;
2145         }
2146
2147         /* create "lov" entry for compatibility purposes */
2148         sym = class_add_symlinks(LUSTRE_LOV_NAME, true);
2149         if (IS_ERR(sym)) {
2150                 rc = PTR_ERR(sym);
2151                 /* does real "lov" already exist ? */
2152                 if (rc == -EEXIST)
2153                         rc = 0;
2154         }
2155
2156         return rc;
2157 }
2158
2159 static void __exit lod_exit(void)
2160 {
2161         struct obd_type *sym = class_search_type(LUSTRE_LOV_NAME);
2162
2163         /* if this was never fully initialized by the lov layer
2164          * then we are responsible for freeing this obd_type
2165          */
2166         if (sym) {
2167                 /* final put if we manage this obd type */
2168                 if (sym->typ_sym_filter)
2169                         kobject_put(&sym->typ_kobj);
2170                 /* put reference taken by class_search_type */
2171                 kobject_put(&sym->typ_kobj);
2172         }
2173
2174         class_unregister_type(LUSTRE_LOD_NAME);
2175         lu_kmem_fini(lod_caches);
2176 }
2177
2178 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2179 MODULE_DESCRIPTION("Lustre Logical Object Device ("LUSTRE_LOD_NAME")");
2180 MODULE_VERSION(LUSTRE_VERSION_STRING);
2181 MODULE_LICENSE("GPL");
2182
2183 module_init(lod_init);
2184 module_exit(lod_exit);