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  *
  31  * lustre/lod/lod_dev.c
  32  *
  33  * Lustre Logical Object Device
  34  *
  35  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  36  * Author: Mikhail Pershin <mike.pershin@intel.com>
  37  */
  38 /**
  39  * The Logical Object Device (LOD) layer manages access to striped
  40  * objects (both regular files and directories). It implements the DT
  41  * device and object APIs and is responsible for creating, storing,
  42  * and loading striping information as an extended attribute of the
  43  * underlying OSD object. LOD is the server side analog of the LOV and
  44  * LMV layers on the client side.
  45  *
  46  * Metadata LU object stack (layers of the same compound LU object,
  47  * all have the same FID):
  48  *
  49  *        MDT
  50  *         |      MD API
  51  *        MDD
  52  *         |      DT API
  53  *        LOD
  54  *       /   \    DT API
  55  *     OSD   OSP
  56  *
  57  * During LOD object initialization the localness or remoteness of the
  58  * object FID dictates the choice between OSD and OSP.
  59  *
  60  * An LOD object (file or directory) with N stripes (each has a
  61  * different FID):
  62  *
  63  *          LOD
  64  *           |
  65  *   +---+---+---+...+
  66  *   |   |   |   |   |
  67  *   S0  S1  S2  S3  S(N-1)  OS[DP] objects, seen as DT objects by LOD
  68  *
  69  * When upper layers must access an object's stripes (which are
  70  * themselves OST or MDT LU objects) LOD finds these objects by their
  71  * FIDs and stores them as an array of DT object pointers on the
  72  * object. Declarations and operations on LOD objects are received by
  73  * LOD (as DT object operations) and performed on the underlying
  74  * OS[DP] object and (as needed) on the stripes. From the perspective
  75  * of LOD, a stripe-less file (created by mknod() or open with
  76  * O_LOV_DELAY_CREATE) is an object which does not yet have stripes,
  77  * while a non-striped directory (created by mkdir()) is an object
  78  * which will never have stripes.
  79  *
  80  * The LOD layer also implements a small subset of the OBD device API
  81  * to support MDT stack initialization and finalization (an MDD device
  82  * connects and disconnects itself to and from the underlying LOD
  83  * device), and pool management. In turn LOD uses the OBD device API
  84  * to connect it self to the underlying OSD, and to connect itself to
  85  * OSP devices representing the MDTs and OSTs that bear the stripes of
  86  * its objects.
  87  */
  88
  89 #define DEBUG_SUBSYSTEM S_MDS
  90
  91 #include <linux/kthread.h>
  92 #include <obd_class.h>
  93 #include <md_object.h>
  94 #include <lustre_fid.h>
  95 #include <uapi/linux/lustre/lustre_param.h>
  96 #include <lustre_update.h>
  97 #include <lustre_log.h>
  98 #include <lustre_lmv.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 /**
1233  * Implementation of lu_device_operations::ldo_fid_alloc() for LOD
1234  *
1235  * Find corresponding device by passed parent and name, and allocate FID from
1236  * there.
1237  *
1238  * see include/lu_object.h for the details.
1239  */
1240 static int lod_fid_alloc(const struct lu_env *env, struct lu_device *d,
1241                          struct lu_fid *fid, struct lu_object *parent,
1242                          const struct lu_name *name)
1243 {
1244         struct lod_device *lod = lu2lod_dev(d);
1245         struct lod_object *lo = lu2lod_obj(parent);
1246         struct dt_device *next;
1247         int rc;
1248
1249         ENTRY;
1250
1251         /* if @parent is remote, we don't know whether its layout was changed,
1252          * always reload layout.
1253          */
1254         if (lu_object_remote(parent))
1255                 lod_striping_free(env, lo);
1256
1257         rc = lod_striping_load(env, lo);
1258         if (rc)
1259                 RETURN(rc);
1260
1261         if (lo->ldo_dir_stripe_count > 0 && name) {
1262                 struct dt_object *stripe;
1263                 int idx;
1264
1265                 idx = __lmv_name_to_stripe_index(lo->ldo_dir_hash_type,
1266                                                  lo->ldo_dir_stripe_count,
1267                                                  lo->ldo_dir_migrate_hash,
1268                                                  lo->ldo_dir_migrate_offset,
1269                                                  name->ln_name,
1270                                                  name->ln_namelen, true);
1271                 if (idx < 0)
1272                         RETURN(idx);
1273
1274                 stripe = lo->ldo_stripe[idx];
1275                 if (!stripe || !dt_object_exists(stripe))
1276                         RETURN(-ENODEV);
1277
1278                 next = lu2dt_dev(stripe->do_lu.lo_dev);
1279         } else {
1280                 next = lod->lod_child;
1281         }
1282
1283         rc = dt_fid_alloc(env, next, fid, parent, name);
1284
1285         RETURN(rc);
1286 }
1287
1288 const struct lu_device_operations lod_lu_ops = {
1289         .ldo_object_alloc       = lod_object_alloc,
1290         .ldo_process_config     = lod_process_config,
1291         .ldo_recovery_complete  = lod_recovery_complete,
1292         .ldo_prepare            = lod_prepare,
1293         .ldo_fid_alloc          = lod_fid_alloc,
1294 };
1295
1296 /**
1297  * Implementation of dt_device_operations::dt_root_get() for LOD
1298  *
1299  * see include/dt_object.h for the details.
1300  */
1301 static int lod_root_get(const struct lu_env *env,
1302                         struct dt_device *dev, struct lu_fid *f)
1303 {
1304         return dt_root_get(env, dt2lod_dev(dev)->lod_child, f);
1305 }
1306
1307 static void lod_statfs_sum(struct obd_statfs *sfs,
1308                              struct obd_statfs *ost_sfs, int *bs)
1309 {
1310         while (ost_sfs->os_bsize < *bs) {
1311                 *bs >>= 1;
1312                 sfs->os_bsize >>= 1;
1313                 sfs->os_bavail <<= 1;
1314                 sfs->os_blocks <<= 1;
1315                 sfs->os_bfree <<= 1;
1316                 sfs->os_granted <<= 1;
1317         }
1318         while (ost_sfs->os_bsize > *bs) {
1319                 ost_sfs->os_bsize >>= 1;
1320                 ost_sfs->os_bavail <<= 1;
1321                 ost_sfs->os_blocks <<= 1;
1322                 ost_sfs->os_bfree <<= 1;
1323                 ost_sfs->os_granted <<= 1;
1324         }
1325         sfs->os_bavail += ost_sfs->os_bavail;
1326         sfs->os_blocks += ost_sfs->os_blocks;
1327         sfs->os_bfree += ost_sfs->os_bfree;
1328         sfs->os_granted += ost_sfs->os_granted;
1329 }
1330
1331 /**
1332  * Implementation of dt_device_operations::dt_statfs() for LOD
1333  *
1334  * see include/dt_object.h for the details.
1335  */
1336 static int lod_statfs(const struct lu_env *env, struct dt_device *dev,
1337                       struct obd_statfs *sfs, struct obd_statfs_info *info)
1338 {
1339         struct lod_device *lod = dt2lod_dev(dev);
1340         struct lu_tgt_desc *tgt;
1341         struct obd_statfs ost_sfs;
1342         u64 ost_files = 0;
1343         u64 ost_ffree = 0;
1344         int rc, bs;
1345
1346         rc = dt_statfs(env, dt2lod_dev(dev)->lod_child, sfs);
1347         if (rc)
1348                 GOTO(out, rc);
1349
1350         bs = sfs->os_bsize;
1351
1352         sfs->os_bavail = 0;
1353         sfs->os_blocks = 0;
1354         sfs->os_bfree = 0;
1355         sfs->os_granted = 0;
1356
1357         lod_getref(&lod->lod_mdt_descs);
1358         lod_foreach_mdt(lod, tgt) {
1359                 rc = dt_statfs(env, tgt->ltd_tgt, &ost_sfs);
1360                 /* ignore errors */
1361                 if (rc)
1362                         continue;
1363                 sfs->os_files += ost_sfs.os_files;
1364                 sfs->os_ffree += ost_sfs.os_ffree;
1365                 lod_statfs_sum(sfs, &ost_sfs, &bs);
1366         }
1367         lod_putref(lod, &lod->lod_mdt_descs);
1368
1369         /*
1370          * at some point we can check whether DoM is enabled and
1371          * decide how to account MDT space. for simplicity let's
1372          * just fallback to pre-DoM policy if any OST is alive
1373          */
1374         lod_getref(&lod->lod_ost_descs);
1375         lod_foreach_ost(lod, tgt) {
1376                 rc = dt_statfs(env, tgt->ltd_tgt, &ost_sfs);
1377                 /* ignore errors */
1378                 if (rc || ost_sfs.os_bsize == 0)
1379                         continue;
1380                 if (!ost_files) {
1381                         /*
1382                          * if only MDTs with DoM then report only MDT blocks,
1383                          * otherwise show only OST blocks, and DoM is "free"
1384                          */
1385                         sfs->os_bavail = 0;
1386                         sfs->os_blocks = 0;
1387                         sfs->os_bfree = 0;
1388                         sfs->os_granted = 0;
1389                 }
1390                 ost_files += ost_sfs.os_files;
1391                 ost_ffree += ost_sfs.os_ffree;
1392                 ost_sfs.os_bavail += ost_sfs.os_granted;
1393                 lod_statfs_sum(sfs, &ost_sfs, &bs);
1394                 LASSERTF(bs == ost_sfs.os_bsize, "%d != %d\n",
1395                         (int)sfs->os_bsize, (int)ost_sfs.os_bsize);
1396         }
1397         lod_putref(lod, &lod->lod_ost_descs);
1398         sfs->os_state |= OS_STATFS_SUM;
1399
1400         /* If we have _some_ OSTs, but don't have as many free objects on the
1401          * OSTs as inodes on the MDTs, reduce the reported number of inodes
1402          * to compensate, so that the "inodes in use" number is correct.
1403          * This should be kept in sync with ll_statfs_internal().
1404          */
1405         if (ost_files && ost_ffree < sfs->os_ffree) {
1406                 sfs->os_files = (sfs->os_files - sfs->os_ffree) + ost_ffree;
1407                 sfs->os_ffree = ost_ffree;
1408         }
1409
1410         /* a single successful statfs should be enough */
1411         rc = 0;
1412
1413 out:
1414         RETURN(rc);
1415 }
1416
1417 /**
1418  * Implementation of dt_device_operations::dt_trans_create() for LOD
1419  *
1420  * Creates a transaction using local (to this node) OSD.
1421  *
1422  * see include/dt_object.h for the details.
1423  */
1424 static struct thandle *lod_trans_create(const struct lu_env *env,
1425                                         struct dt_device *dt)
1426 {
1427         struct thandle *th;
1428
1429         th = top_trans_create(env, dt2lod_dev(dt)->lod_child);
1430         if (IS_ERR(th))
1431                 return th;
1432
1433         th->th_dev = dt;
1434
1435         return th;
1436 }
1437
1438 /**
1439  * Implementation of dt_device_operations::dt_trans_start() for LOD
1440  *
1441  * Starts the set of local transactions using the targets involved
1442  * in declare phase. Initial support for the distributed transactions.
1443  *
1444  * see include/dt_object.h for the details.
1445  */
1446 static int lod_trans_start(const struct lu_env *env, struct dt_device *dt,
1447                            struct thandle *th)
1448 {
1449         return top_trans_start(env, dt2lod_dev(dt)->lod_child, th);
1450 }
1451
1452 static int lod_trans_cb_add(struct thandle *th,
1453                             struct dt_txn_commit_cb *dcb)
1454 {
1455         struct top_thandle      *top_th = container_of(th, struct top_thandle,
1456                                                        tt_super);
1457         return dt_trans_cb_add(top_th->tt_master_sub_thandle, dcb);
1458 }
1459
1460 /**
1461  * add noop update to the update records
1462  *
1463  * Add noop updates to the update records, which is only used in
1464  * test right now.
1465  *
1466  * \param[in] env       execution environment
1467  * \param[in] dt        dt device of lod
1468  * \param[in] th        thandle
1469  * \param[in] count     the count of update records to be added.
1470  *
1471  * \retval              0 if adding succeeds.
1472  * \retval              negative errno if adding fails.
1473  */
1474 static int lod_add_noop_records(const struct lu_env *env,
1475                                 struct dt_device *dt, struct thandle *th,
1476                                 int count)
1477 {
1478         struct top_thandle *top_th;
1479         struct lu_fid *fid = &lod_env_info(env)->lti_fid;
1480         int i;
1481         int rc = 0;
1482
1483         top_th = container_of(th, struct top_thandle, tt_super);
1484         if (!top_th->tt_multiple_thandle)
1485                 return 0;
1486
1487         fid_zero(fid);
1488         for (i = 0; i < count; i++) {
1489                 rc = update_record_pack(noop, th, fid);
1490                 if (rc < 0)
1491                         return rc;
1492         }
1493         return rc;
1494 }
1495
1496 /**
1497  * Implementation of dt_device_operations::dt_trans_stop() for LOD
1498  *
1499  * Stops the set of local transactions using the targets involved
1500  * in declare phase. Initial support for the distributed transactions.
1501  *
1502  * see include/dt_object.h for the details.
1503  */
1504 static int lod_trans_stop(const struct lu_env *env, struct dt_device *dt,
1505                           struct thandle *th)
1506 {
1507         if (OBD_FAIL_CHECK(OBD_FAIL_SPLIT_UPDATE_REC)) {
1508                 int rc;
1509
1510                 rc = lod_add_noop_records(env, dt, th, 5000);
1511                 if (rc < 0)
1512                         RETURN(rc);
1513         }
1514         return top_trans_stop(env, dt2lod_dev(dt)->lod_child, th);
1515 }
1516
1517 /**
1518  * Implementation of dt_device_operations::dt_conf_get() for LOD
1519  *
1520  * Currently returns the configuration provided by the local OSD.
1521  *
1522  * see include/dt_object.h for the details.
1523  */
1524 static void lod_conf_get(const struct lu_env *env,
1525                          const struct dt_device *dev,
1526                          struct dt_device_param *param)
1527 {
1528         dt_conf_get(env, dt2lod_dev((struct dt_device *)dev)->lod_child, param);
1529 }
1530
1531 /**
1532  * Implementation of dt_device_operations::dt_sync() for LOD
1533  *
1534  * Syncs all known OST targets. Very very expensive and used
1535  * rarely by LFSCK now. Should not be used in general.
1536  *
1537  * see include/dt_object.h for the details.
1538  */
1539 static int lod_sync(const struct lu_env *env, struct dt_device *dev)
1540 {
1541         struct lod_device *lod = dt2lod_dev(dev);
1542         struct lu_tgt_desc *tgt;
1543         int rc = 0;
1544
1545         ENTRY;
1546
1547         lod_getref(&lod->lod_ost_descs);
1548         lod_foreach_ost(lod, tgt) {
1549                 if (!tgt->ltd_active)
1550                         continue;
1551                 rc = dt_sync(env, tgt->ltd_tgt);
1552                 if (rc) {
1553                         if (rc != -ENOTCONN) {
1554                                 CERROR("%s: can't sync ost %u: rc = %d\n",
1555                                        lod2obd(lod)->obd_name, tgt->ltd_index,
1556                                        rc);
1557                                 break;
1558                         }
1559                         rc = 0;
1560                 }
1561         }
1562         lod_putref(lod, &lod->lod_ost_descs);
1563
1564         if (rc)
1565                 RETURN(rc);
1566
1567         lod_getref(&lod->lod_mdt_descs);
1568         lod_foreach_mdt(lod, tgt) {
1569                 if (!tgt->ltd_active)
1570                         continue;
1571                 rc = dt_sync(env, tgt->ltd_tgt);
1572                 if (rc) {
1573                         if (rc != -ENOTCONN) {
1574                                 CERROR("%s: can't sync mdt %u: rc = %d\n",
1575                                        lod2obd(lod)->obd_name, tgt->ltd_index,
1576                                        rc);
1577                                 break;
1578                         }
1579                         rc = 0;
1580                 }
1581         }
1582         lod_putref(lod, &lod->lod_mdt_descs);
1583
1584         if (rc == 0)
1585                 rc = dt_sync(env, lod->lod_child);
1586
1587         RETURN(rc);
1588 }
1589
1590 /**
1591  * Implementation of dt_device_operations::dt_ro() for LOD
1592  *
1593  * Turns local OSD read-only, used for the testing only.
1594  *
1595  * see include/dt_object.h for the details.
1596  */
1597 static int lod_ro(const struct lu_env *env, struct dt_device *dev)
1598 {
1599         return dt_ro(env, dt2lod_dev(dev)->lod_child);
1600 }
1601
1602 /**
1603  * Implementation of dt_device_operations::dt_commit_async() for LOD
1604  *
1605  * Asks local OSD to commit sooner.
1606  *
1607  * see include/dt_object.h for the details.
1608  */
1609 static int lod_commit_async(const struct lu_env *env, struct dt_device *dev)
1610 {
1611         return dt_commit_async(env, dt2lod_dev(dev)->lod_child);
1612 }
1613
1614 static const struct dt_device_operations lod_dt_ops = {
1615         .dt_root_get         = lod_root_get,
1616         .dt_statfs           = lod_statfs,
1617         .dt_trans_create     = lod_trans_create,
1618         .dt_trans_start      = lod_trans_start,
1619         .dt_trans_stop       = lod_trans_stop,
1620         .dt_conf_get         = lod_conf_get,
1621         .dt_sync             = lod_sync,
1622         .dt_ro               = lod_ro,
1623         .dt_commit_async     = lod_commit_async,
1624         .dt_trans_cb_add     = lod_trans_cb_add,
1625 };
1626
1627 /**
1628  * Connect to a local OSD.
1629  *
1630  * Used to connect to the local OSD at mount. OSD name is taken from the
1631  * configuration command passed. This connection is used to identify LU
1632  * site and pin the OSD from early removal.
1633  *
1634  * \param[in] env               LU environment provided by the caller
1635  * \param[in] lod               lod device
1636  * \param[in] cfg               configuration command to apply
1637  *
1638  * \retval 0                    on success
1639  * \retval negative             negated errno on error
1640  **/
1641 static int lod_connect_to_osd(const struct lu_env *env, struct lod_device *lod,
1642                               struct lustre_cfg *cfg)
1643 {
1644         struct obd_connect_data *data = NULL;
1645         struct obd_device *obd;
1646         char *nextdev = NULL, *p, *s;
1647         int rc, len = 0;
1648
1649         ENTRY;
1650
1651         LASSERT(cfg);
1652         LASSERT(lod->lod_child_exp == NULL);
1653
1654         /*
1655          * compatibility hack: we still use old config logs
1656          * which specify LOV, but we need to learn underlying
1657          * OSD device, which is supposed to be:
1658          *  <fsname>-MDTxxxx-osd
1659          *
1660          * 2.x MGS generates lines like the following:
1661          *   #03 (176)lov_setup 0:lustre-MDT0000-mdtlov  1:(struct lov_desc)
1662          * 1.8 MGS generates lines like the following:
1663          *   #03 (168)lov_setup 0:lustre-mdtlov  1:(struct lov_desc)
1664          *
1665          * we use "-MDT" to differentiate 2.x from 1.8
1666          */
1667         p = lustre_cfg_string(cfg, 0);
1668         if (p && strstr(p, "-mdtlov")) {
1669                 len = strlen(p) + 6;
1670                 OBD_ALLOC(nextdev, len);
1671                 if (!nextdev)
1672                         GOTO(out, rc = -ENOMEM);
1673
1674                 strcpy(nextdev, p);
1675                 s = strstr(nextdev, "-mdtlov");
1676                 if (unlikely(!s)) {
1677                         CERROR("%s: unable to parse device name: rc = %d\n",
1678                                lustre_cfg_string(cfg, 0), -EINVAL);
1679                         GOTO(out, rc = -EINVAL);
1680                 }
1681
1682                 if (strstr(nextdev, "-MDT")) {
1683                         /* 2.x config */
1684                         strcpy(s, "-osd");
1685                 } else {
1686                         /* 1.8 config */
1687                         strcpy(s, "-MDT0000-osd");
1688                 }
1689         } else {
1690                 CERROR("%s: unable to parse device name: rc = %d\n",
1691                        lustre_cfg_string(cfg, 0), -EINVAL);
1692                 GOTO(out, rc = -EINVAL);
1693         }
1694
1695         OBD_ALLOC_PTR(data);
1696         if (!data)
1697                 GOTO(out, rc = -ENOMEM);
1698
1699         obd = class_name2obd(nextdev);
1700         if (!obd) {
1701                 CERROR("%s: can not locate next device: rc = %d\n",
1702                        nextdev, -ENOTCONN);
1703                 GOTO(out, rc = -ENOTCONN);
1704         }
1705
1706         data->ocd_connect_flags = OBD_CONNECT_VERSION;
1707         data->ocd_version = LUSTRE_VERSION_CODE;
1708
1709         rc = obd_connect(env, &lod->lod_child_exp, obd, &obd->obd_uuid,
1710                          data, NULL);
1711         if (rc) {
1712                 CERROR("%s: cannot connect to next dev: rc = %d\n",
1713                        nextdev, rc);
1714                 GOTO(out, rc);
1715         }
1716
1717         lod->lod_dt_dev.dd_lu_dev.ld_site =
1718                 lod->lod_child_exp->exp_obd->obd_lu_dev->ld_site;
1719         LASSERT(lod->lod_dt_dev.dd_lu_dev.ld_site);
1720         lod->lod_child = lu2dt_dev(lod->lod_child_exp->exp_obd->obd_lu_dev);
1721
1722 out:
1723         if (data)
1724                 OBD_FREE_PTR(data);
1725         if (nextdev)
1726                 OBD_FREE(nextdev, len);
1727         RETURN(rc);
1728 }
1729
1730 static int lod_lsfs_init(const struct lu_env *env, struct lod_device *d)
1731 {
1732         struct obd_statfs sfs;
1733         int rc;
1734
1735         rc = dt_statfs(env, d->lod_child, &sfs);
1736         if (rc) {
1737                 CDEBUG(D_LAYOUT, "%s: failed to get OSD statfs, rc = %d\n",
1738                        lod2obd(d)->obd_name, rc);
1739                 return rc;
1740         }
1741
1742         /* udpate local OSD cached statfs data */
1743         spin_lock_init(&d->lod_lsfs_lock);
1744         d->lod_lsfs_age = ktime_get_seconds();
1745         d->lod_lsfs_total_mb = (sfs.os_blocks * sfs.os_bsize) >> 20;
1746         d->lod_lsfs_free_mb = (sfs.os_bfree * sfs.os_bsize) >> 20;
1747         return 0;
1748 }
1749
1750 /**
1751  * Initialize LOD device at setup.
1752  *
1753  * Initializes the given LOD device using the original configuration command.
1754  * The function initiates a connection to the local OSD and initializes few
1755  * internal structures like pools, target tables, etc.
1756  *
1757  * \param[in] env               LU environment provided by the caller
1758  * \param[in] lod               lod device
1759  * \param[in] ldt               not used
1760  * \param[in] cfg               configuration command
1761  *
1762  * \retval 0                    on success
1763  * \retval negative             negated errno on error
1764  **/
1765 static int lod_init0(const struct lu_env *env, struct lod_device *lod,
1766                      struct lu_device_type *ldt, struct lustre_cfg *cfg)
1767 {
1768         struct dt_device_param ddp;
1769         struct obd_device *obd;
1770         int rc;
1771
1772         ENTRY;
1773
1774         obd = class_name2obd(lustre_cfg_string(cfg, 0));
1775         if (!obd) {
1776                 rc = -ENODEV;
1777                 CERROR("Cannot find obd with name '%s': rc = %d\n",
1778                        lustre_cfg_string(cfg, 0), rc);
1779                 RETURN(rc);
1780         }
1781
1782         obd->obd_lu_dev = &lod->lod_dt_dev.dd_lu_dev;
1783         lod->lod_dt_dev.dd_lu_dev.ld_obd = obd;
1784         lod->lod_dt_dev.dd_lu_dev.ld_ops = &lod_lu_ops;
1785         lod->lod_dt_dev.dd_ops = &lod_dt_ops;
1786
1787         rc = lod_connect_to_osd(env, lod, cfg);
1788         if (rc)
1789                 RETURN(rc);
1790
1791         dt_conf_get(env, &lod->lod_dt_dev, &ddp);
1792         lod->lod_osd_max_easize = ddp.ddp_max_ea_size;
1793         lod->lod_dom_stripesize_max_kb = (1ULL << 10); /* 1Mb is default */
1794
1795         /* initialize local statfs cached values */
1796         rc = lod_lsfs_init(env, lod);
1797         if (rc)
1798                 GOTO(out_disconnect, rc);
1799
1800         /* default threshold as half of total space, in MiB */
1801         lod->lod_dom_threshold_free_mb = lod->lod_lsfs_total_mb / 2;
1802         /* set default DoM stripe size based on free space amount */
1803         lod_dom_stripesize_recalc(lod);
1804
1805         /* setup obd to be used with old lov code */
1806         rc = lod_pools_init(lod, cfg);
1807         if (rc)
1808                 GOTO(out_disconnect, rc);
1809
1810         rc = lod_procfs_init(lod);
1811         if (rc)
1812                 GOTO(out_pools, rc);
1813
1814         spin_lock_init(&lod->lod_lock);
1815         spin_lock_init(&lod->lod_connects_lock);
1816         lu_tgt_descs_init(&lod->lod_mdt_descs, true);
1817         lu_tgt_descs_init(&lod->lod_ost_descs, false);
1818         lu_qos_rr_init(&lod->lod_mdt_descs.ltd_qos.lq_rr);
1819         lu_qos_rr_init(&lod->lod_ost_descs.ltd_qos.lq_rr);
1820
1821         RETURN(0);
1822
1823 out_pools:
1824         lod_pools_fini(lod);
1825 out_disconnect:
1826         obd_disconnect(lod->lod_child_exp);
1827         RETURN(rc);
1828 }
1829
1830 /**
1831  * Implementation of lu_device_type_operations::ldto_device_free() for LOD
1832  *
1833  * Releases the memory allocated for LOD device.
1834  *
1835  * see include/lu_object.h for the details.
1836  */
1837 static struct lu_device *lod_device_free(const struct lu_env *env,
1838                                          struct lu_device *lu)
1839 {
1840         struct lod_device *lod = lu2lod_dev(lu);
1841         struct lu_device  *next = &lod->lod_child->dd_lu_dev;
1842
1843         ENTRY;
1844
1845         if (atomic_read(&lu->ld_site->ls_obj_hash.nelems)) {
1846                 lu_site_print(env, lu->ld_site, &lu->ld_ref, D_ERROR,
1847                               lu_cdebug_printer);
1848         }
1849         LASSERTF(atomic_read(&lu->ld_ref) == 0, "lu is %p\n", lu);
1850         dt_device_fini(&lod->lod_dt_dev);
1851         OBD_FREE_PTR(lod);
1852         RETURN(next);
1853 }
1854
1855 /**
1856  * Implementation of lu_device_type_operations::ldto_device_alloc() for LOD
1857  *
1858  * Allocates LOD device and calls the helpers to initialize it.
1859  *
1860  * see include/lu_object.h for the details.
1861  */
1862 static struct lu_device *lod_device_alloc(const struct lu_env *env,
1863                                           struct lu_device_type *type,
1864                                           struct lustre_cfg *lcfg)
1865 {
1866         struct lod_device *lod;
1867         struct lu_device *lu_dev;
1868
1869         OBD_ALLOC_PTR(lod);
1870         if (!lod) {
1871                 lu_dev = ERR_PTR(-ENOMEM);
1872         } else {
1873                 int rc;
1874
1875                 lu_dev = lod2lu_dev(lod);
1876                 dt_device_init(&lod->lod_dt_dev, type);
1877                 rc = lod_init0(env, lod, type, lcfg);
1878                 if (rc != 0) {
1879                         lod_device_free(env, lu_dev);
1880                         lu_dev = ERR_PTR(rc);
1881                 }
1882         }
1883
1884         return lu_dev;
1885 }
1886
1887 static void lod_avoid_guide_fini(struct lod_avoid_guide *lag)
1888 {
1889         if (lag->lag_oss_avoid_array)
1890                 OBD_FREE_PTR_ARRAY(lag->lag_oss_avoid_array,
1891                                    lag->lag_oaa_size);
1892         bitmap_free(lag->lag_ost_avoid_bitmap);
1893 }
1894
1895 /**
1896  * Implementation of lu_device_type_operations::ldto_device_fini() for LOD
1897  *
1898  * Releases the internal resources used by LOD device.
1899  *
1900  * see include/lu_object.h for the details.
1901  */
1902 static struct lu_device *lod_device_fini(const struct lu_env *env,
1903                                          struct lu_device *d)
1904 {
1905         struct lod_device *lod = lu2lod_dev(d);
1906         int rc;
1907
1908         ENTRY;
1909
1910         lod_pools_fini(lod);
1911
1912         lod_procfs_fini(lod);
1913
1914         rc = lod_fini_tgt(env, lod, &lod->lod_ost_descs);
1915         if (rc)
1916                 CERROR("%s: can not fini ost descriptors: rc =  %d\n",
1917                         lod2obd(lod)->obd_name, rc);
1918
1919         rc = lod_fini_tgt(env, lod, &lod->lod_mdt_descs);
1920         if (rc)
1921                 CERROR("%s: can not fini mdt descriptors: rc =  %d\n",
1922                         lod2obd(lod)->obd_name, rc);
1923
1924         RETURN(NULL);
1925 }
1926
1927 /**
1928  * Implementation of obd_ops::o_connect() for LOD
1929  *
1930  * Used to track all the users of this specific LOD device,
1931  * so the device stays up until the last user disconnected.
1932  *
1933  * \param[in] env               LU environment provided by the caller
1934  * \param[out] exp              export the caller will be using to access LOD
1935  * \param[in] obd               OBD device representing LOD device
1936  * \param[in] cluuid            unique identifier of the caller
1937  * \param[in] data              not used
1938  * \param[in] localdata         not used
1939  *
1940  * \retval 0                    on success
1941  * \retval negative             negated errno on error
1942  **/
1943 static int lod_obd_connect(const struct lu_env *env, struct obd_export **exp,
1944                            struct obd_device *obd, struct obd_uuid *cluuid,
1945                            struct obd_connect_data *data, void *localdata)
1946 {
1947         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
1948         struct lustre_handle conn;
1949         int rc;
1950
1951         ENTRY;
1952
1953         CDEBUG(D_CONFIG, "connect #%d\n", lod->lod_connects);
1954
1955         rc = class_connect(&conn, obd, cluuid);
1956         if (rc)
1957                 RETURN(rc);
1958
1959         *exp = class_conn2export(&conn);
1960
1961         spin_lock(&lod->lod_connects_lock);
1962         lod->lod_connects++;
1963         /* at the moment we expect the only user */
1964         LASSERT(lod->lod_connects == 1);
1965         spin_unlock(&lod->lod_connects_lock);
1966
1967         RETURN(0);
1968 }
1969
1970 /**
1971  *
1972  * Implementation of obd_ops::o_disconnect() for LOD
1973  *
1974  * When the caller doesn't need to use this LOD instance, it calls
1975  * obd_disconnect() and LOD releases corresponding export/reference count.
1976  * Once all the users gone, LOD device is released.
1977  *
1978  * \param[in] exp               export provided to the caller in obd_connect()
1979  *
1980  * \retval 0                    on success
1981  * \retval negative             negated errno on error
1982  **/
1983 static int lod_obd_disconnect(struct obd_export *exp)
1984 {
1985         struct obd_device *obd = exp->exp_obd;
1986         struct lod_device *lod = lu2lod_dev(obd->obd_lu_dev);
1987         int rc, release = 0;
1988
1989         ENTRY;
1990
1991         /* Only disconnect the underlying layers on the final disconnect. */
1992         spin_lock(&lod->lod_connects_lock);
1993         lod->lod_connects--;
1994         if (lod->lod_connects != 0) {
1995                 /* why should there be more than 1 connect? */
1996                 spin_unlock(&lod->lod_connects_lock);
1997                 CERROR("%s: disconnect #%d\n", exp->exp_obd->obd_name,
1998                        lod->lod_connects);
1999                 goto out;
2000         }
2001         spin_unlock(&lod->lod_connects_lock);
2002
2003         /* the last user of lod has gone, let's release the device */
2004         release = 1;
2005
2006 out:
2007         rc = class_disconnect(exp); /* bz 9811 */
2008
2009         if (rc == 0 && release)
2010                 class_manual_cleanup(obd);
2011         RETURN(rc);
2012 }
2013
2014 LU_KEY_INIT(lod, struct lod_thread_info);
2015
2016 static void lod_key_fini(const struct lu_context *ctx,
2017                 struct lu_context_key *key, void *data)
2018 {
2019         struct lod_thread_info *info = data;
2020         struct lod_layout_component *lds =
2021                                 info->lti_def_striping.lds_def_comp_entries;
2022
2023         /*
2024          * allocated in lod_get_lov_ea
2025          * XXX: this is overload, a tread may have such store but used only
2026          * once. Probably better would be pool of such stores per LOD.
2027          */
2028         if (info->lti_ea_store) {
2029                 OBD_FREE_LARGE(info->lti_ea_store, info->lti_ea_store_size);
2030                 info->lti_ea_store = NULL;
2031                 info->lti_ea_store_size = 0;
2032         }
2033         lu_buf_free(&info->lti_linkea_buf);
2034
2035         if (lds)
2036                 lod_free_def_comp_entries(&info->lti_def_striping);
2037
2038         if (info->lti_comp_size > 0)
2039                 OBD_FREE_PTR_ARRAY(info->lti_comp_idx,
2040                                    info->lti_comp_size);
2041
2042         lod_avoid_guide_fini(&info->lti_avoid);
2043
2044         OBD_FREE_PTR(info);
2045 }
2046
2047 /* context key: lod_thread_key */
2048 LU_CONTEXT_KEY_DEFINE(lod, LCT_MD_THREAD);
2049
2050 LU_TYPE_INIT_FINI(lod, &lod_thread_key);
2051
2052 static const struct lu_device_type_operations lod_device_type_ops = {
2053         .ldto_init              = lod_type_init,
2054         .ldto_fini              = lod_type_fini,
2055
2056         .ldto_start             = lod_type_start,
2057         .ldto_stop              = lod_type_stop,
2058
2059         .ldto_device_alloc      = lod_device_alloc,
2060         .ldto_device_free       = lod_device_free,
2061
2062         .ldto_device_fini       = lod_device_fini
2063 };
2064
2065 static struct lu_device_type lod_device_type = {
2066         .ldt_tags     = LU_DEVICE_DT,
2067         .ldt_name     = LUSTRE_LOD_NAME,
2068         .ldt_ops      = &lod_device_type_ops,
2069         .ldt_ctx_tags = LCT_MD_THREAD,
2070 };
2071
2072 /**
2073  * Implementation of obd_ops::o_get_info() for LOD
2074  *
2075  * Currently, there is only one supported key: KEY_OSP_CONNECTED , to provide
2076  * the caller binary status whether LOD has seen connection to any OST target.
2077  * It will also check if the MDT update log context being initialized (if
2078  * needed).
2079  *
2080  * \param[in] env               LU environment provided by the caller
2081  * \param[in] exp               export of the caller
2082  * \param[in] keylen            len of the key
2083  * \param[in] key               the key
2084  * \param[in] vallen            not used
2085  * \param[in] val               not used
2086  *
2087  * \retval                      0 if a connection was seen
2088  * \retval                      -EAGAIN if LOD isn't running yet or no
2089  *                              connection has been seen yet
2090  * \retval                      -EINVAL if not supported key is requested
2091  **/
2092 static int lod_obd_get_info(const struct lu_env *env, struct obd_export *exp,
2093                             u32 keylen, void *key, u32 *vallen, void *val)
2094 {
2095         int rc = -EINVAL;
2096
2097         if (KEY_IS(KEY_OSP_CONNECTED)) {
2098                 struct obd_device *obd = exp->exp_obd;
2099                 struct lod_device *d;
2100                 struct lod_tgt_desc *tgt;
2101                 int rc = 1;
2102
2103                 if (!obd->obd_set_up || obd->obd_stopping)
2104                         RETURN(-EAGAIN);
2105
2106                 d = lu2lod_dev(obd->obd_lu_dev);
2107                 lod_getref(&d->lod_ost_descs);
2108                 lod_foreach_ost(d, tgt) {
2109                         rc = obd_get_info(env, tgt->ltd_exp, keylen, key,
2110                                           vallen, val);
2111                         /* one healthy device is enough */
2112                         if (rc == 0)
2113                                 break;
2114                 }
2115                 lod_putref(d, &d->lod_ost_descs);
2116
2117                 lod_getref(&d->lod_mdt_descs);
2118                 lod_foreach_mdt(d, tgt) {
2119                         struct llog_ctxt *ctxt;
2120
2121                         if (!tgt->ltd_active)
2122                                 continue;
2123
2124                         ctxt = llog_get_context(tgt->ltd_tgt->dd_lu_dev.ld_obd,
2125                                                 LLOG_UPDATELOG_ORIG_CTXT);
2126                         if (!ctxt) {
2127                                 CDEBUG(D_INFO, "%s: %s is not ready.\n",
2128                                        obd->obd_name,
2129                                       tgt->ltd_tgt->dd_lu_dev.ld_obd->obd_name);
2130                                 rc = -EAGAIN;
2131                                 break;
2132                         }
2133                         if (!ctxt->loc_handle) {
2134                                 CDEBUG(D_INFO, "%s: %s is not ready.\n",
2135                                        obd->obd_name,
2136                                       tgt->ltd_tgt->dd_lu_dev.ld_obd->obd_name);
2137                                 rc = -EAGAIN;
2138                                 llog_ctxt_put(ctxt);
2139                                 break;
2140                         }
2141                         llog_ctxt_put(ctxt);
2142                 }
2143                 lod_putref(d, &d->lod_mdt_descs);
2144
2145                 RETURN(rc);
2146         }
2147
2148         RETURN(rc);
2149 }
2150
2151 static int lod_obd_set_info_async(const struct lu_env *env,
2152                                   struct obd_export *exp,
2153                                   u32 keylen, void *key,
2154                                   u32 vallen, void *val,
2155                                   struct ptlrpc_request_set *set)
2156 {
2157         struct obd_device *obd = class_exp2obd(exp);
2158         struct lod_device *d;
2159         struct lod_tgt_desc *tgt;
2160         int no_set = 0;
2161         int rc = 0, rc2;
2162
2163         ENTRY;
2164
2165         if (!set) {
2166                 no_set = 1;
2167                 set = ptlrpc_prep_set();
2168                 if (!set)
2169                         RETURN(-ENOMEM);
2170         }
2171
2172         d = lu2lod_dev(obd->obd_lu_dev);
2173         lod_getref(&d->lod_ost_descs);
2174         lod_foreach_ost(d, tgt) {
2175                 if (!tgt->ltd_active)
2176                         continue;
2177
2178                 rc2 = obd_set_info_async(env, tgt->ltd_exp, keylen, key,
2179                                          vallen, val, set);
2180                 if (rc2 != 0 && rc == 0)
2181                         rc = rc2;
2182         }
2183         lod_putref(d, &d->lod_ost_descs);
2184
2185         lod_getref(&d->lod_mdt_descs);
2186         lod_foreach_mdt(d, tgt) {
2187                 if (!tgt->ltd_active)
2188                         continue;
2189                 rc2 = obd_set_info_async(env, tgt->ltd_exp, keylen, key,
2190                                          vallen, val, set);
2191                 if (rc2 != 0 && rc == 0)
2192                         rc = rc2;
2193         }
2194         lod_putref(d, &d->lod_mdt_descs);
2195
2196
2197         if (no_set) {
2198                 rc2 = ptlrpc_set_wait(env, set);
2199                 if (rc2 == 0 && rc == 0)
2200                         rc = rc2;
2201                 ptlrpc_set_destroy(set);
2202         }
2203         RETURN(rc);
2204 }
2205
2206
2207 #define QMT0_DEV_NAME_LEN (LUSTRE_MAXFSNAME + sizeof("-QMT0000"))
2208 static struct obd_device *obd_find_qmt0(char *obd_name)
2209 {
2210         char qmt_name[QMT0_DEV_NAME_LEN];
2211         struct obd_device *qmt = NULL;
2212
2213         if (!server_name2fsname(obd_name, qmt_name, NULL)) {
2214                 strlcat(qmt_name, "-QMT0000", QMT0_DEV_NAME_LEN);
2215                 qmt = class_name2obd(qmt_name);
2216         }
2217
2218         return qmt;
2219 }
2220
2221 static int lod_pool_new_q(struct obd_device *obd, char *poolname)
2222 {
2223         int err = lod_pool_new(obd, poolname);
2224
2225         if (!err) {
2226                 obd = obd_find_qmt0(obd->obd_name);
2227                 if (obd)
2228                         obd_pool_new(obd, poolname);
2229         }
2230
2231         return err;
2232 }
2233
2234 static int lod_pool_remove_q(struct obd_device *obd, char *poolname,
2235                              char *ostname)
2236 {
2237         int err = lod_pool_remove(obd, poolname, ostname);
2238
2239         if (!err) {
2240                 obd = obd_find_qmt0(obd->obd_name);
2241                 if (obd)
2242                         obd_pool_rem(obd, poolname, ostname);
2243         }
2244
2245         return err;
2246 }
2247
2248 static int lod_pool_add_q(struct obd_device *obd, char *poolname, char *ostname)
2249 {
2250         int err = lod_pool_add(obd, poolname, ostname);
2251
2252         if (!err) {
2253                 obd = obd_find_qmt0(obd->obd_name);
2254                 if (obd)
2255                         obd_pool_add(obd, poolname, ostname);
2256         }
2257
2258         return err;
2259 }
2260
2261 static int lod_pool_del_q(struct obd_device *obd, char *poolname)
2262 {
2263         int err = lod_pool_del(obd, poolname);
2264
2265         if (!err) {
2266                 obd = obd_find_qmt0(obd->obd_name);
2267                 if (obd)
2268                         obd_pool_del(obd, poolname);
2269         }
2270
2271         return err;
2272 }
2273
2274 static const struct obd_ops lod_obd_device_ops = {
2275         .o_owner        = THIS_MODULE,
2276         .o_connect      = lod_obd_connect,
2277         .o_disconnect   = lod_obd_disconnect,
2278         .o_get_info     = lod_obd_get_info,
2279         .o_set_info_async = lod_obd_set_info_async,
2280         .o_pool_new     = lod_pool_new_q,
2281         .o_pool_rem     = lod_pool_remove_q,
2282         .o_pool_add     = lod_pool_add_q,
2283         .o_pool_del     = lod_pool_del_q,
2284 };
2285
2286 static int __init lod_init(void)
2287 {
2288         struct obd_type *sym;
2289         int rc;
2290
2291         rc = lu_kmem_init(lod_caches);
2292         if (rc)
2293                 return rc;
2294
2295         rc = class_register_type(&lod_obd_device_ops, NULL, true,
2296                                  LUSTRE_LOD_NAME, &lod_device_type);
2297         if (rc) {
2298                 lu_kmem_fini(lod_caches);
2299                 return rc;
2300         }
2301
2302         /* create "lov" entry for compatibility purposes */
2303         sym = class_add_symlinks(LUSTRE_LOV_NAME, true);
2304         if (IS_ERR(sym)) {
2305                 rc = PTR_ERR(sym);
2306                 /* does real "lov" already exist ? */
2307                 if (rc == -EEXIST)
2308                         rc = 0;
2309         }
2310
2311         return rc;
2312 }
2313
2314 static void __exit lod_exit(void)
2315 {
2316         struct obd_type *sym = class_search_type(LUSTRE_LOV_NAME);
2317
2318         /* if this was never fully initialized by the lov layer
2319          * then we are responsible for freeing this obd_type
2320          */
2321         if (sym) {
2322                 /* final put if we manage this obd type */
2323                 if (sym->typ_sym_filter)
2324                         kobject_put(&sym->typ_kobj);
2325                 /* put reference taken by class_search_type */
2326                 kobject_put(&sym->typ_kobj);
2327         }
2328
2329         class_unregister_type(LUSTRE_LOD_NAME);
2330         lu_kmem_fini(lod_caches);
2331 }
2332
2333 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2334 MODULE_DESCRIPTION("Lustre Logical Object Device ("LUSTRE_LOD_NAME")");
2335 MODULE_VERSION(LUSTRE_VERSION_STRING);
2336 MODULE_LICENSE("GPL");
2337
2338 module_init(lod_init);
2339 module_exit(lod_exit);