lustre/lfsck/lfsck_layout.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 (c) 2013, Intel Corporation.
  24  */
  25 /*
  26  * lustre/lfsck/lfsck_layout.c
  27  *
  28  * Author: Fan, Yong <fan.yong@intel.com>
  29  */
  30
  31 #ifndef EXPORT_SYMTAB
  32 # define EXPORT_SYMTAB
  33 #endif
  34 #define DEBUG_SUBSYSTEM S_LFSCK
  35
  36 #include <linux/bitops.h>
  37
  38 #include <lustre/lustre_idl.h>
  39 #include <lu_object.h>
  40 #include <dt_object.h>
  41 #include <lustre_linkea.h>
  42 #include <lustre_fid.h>
  43 #include <lustre_lib.h>
  44 #include <lustre_net.h>
  45 #include <lustre/lustre_user.h>
  46 #include <md_object.h>
  47
  48 #include "lfsck_internal.h"
  49
  50 #define LFSCK_LAYOUT_MAGIC              0xB173AE14
  51
  52 static const char lfsck_layout_name[] = "lfsck_layout";
  53
  54 struct lfsck_layout_seq {
  55         struct list_head         lls_list;
  56         __u64                    lls_seq;
  57         __u64                    lls_lastid;
  58         __u64                    lls_lastid_known;
  59         struct dt_object        *lls_lastid_obj;
  60         unsigned int             lls_dirty:1;
  61 };
  62
  63 struct lfsck_layout_slave_data {
  64         /* list for lfsck_layout_seq */
  65         struct list_head         llsd_seq_list;
  66 };
  67
  68 static void lfsck_layout_le_to_cpu(struct lfsck_layout *des,
  69                                    const struct lfsck_layout *src)
  70 {
  71         int i;
  72
  73         des->ll_magic = le32_to_cpu(src->ll_magic);
  74         des->ll_status = le32_to_cpu(src->ll_status);
  75         des->ll_flags = le32_to_cpu(src->ll_flags);
  76         des->ll_success_count = le32_to_cpu(src->ll_success_count);
  77         des->ll_run_time_phase1 = le32_to_cpu(src->ll_run_time_phase1);
  78         des->ll_run_time_phase2 = le32_to_cpu(src->ll_run_time_phase2);
  79         des->ll_time_last_complete = le64_to_cpu(src->ll_time_last_complete);
  80         des->ll_time_latest_start = le64_to_cpu(src->ll_time_latest_start);
  81         des->ll_time_last_checkpoint =
  82                                 le64_to_cpu(src->ll_time_last_checkpoint);
  83         des->ll_pos_latest_start = le64_to_cpu(src->ll_pos_latest_start);
  84         des->ll_pos_last_checkpoint = le64_to_cpu(src->ll_pos_last_checkpoint);
  85         des->ll_pos_first_inconsistent =
  86                         le64_to_cpu(src->ll_pos_first_inconsistent);
  87         des->ll_objs_checked_phase1 = le64_to_cpu(src->ll_objs_checked_phase1);
  88         des->ll_objs_failed_phase1 = le64_to_cpu(src->ll_objs_failed_phase1);
  89         des->ll_objs_checked_phase2 = le64_to_cpu(src->ll_objs_checked_phase2);
  90         des->ll_objs_failed_phase2 = le64_to_cpu(src->ll_objs_failed_phase2);
  91         for (i = 0; i < LLIT_MAX; i++)
  92                 des->ll_objs_repaired[i] =
  93                                 le64_to_cpu(src->ll_objs_repaired[i]);
  94         des->ll_objs_skipped = le64_to_cpu(src->ll_objs_skipped);
  95 }
  96
  97 static void lfsck_layout_cpu_to_le(struct lfsck_layout *des,
  98                                    const struct lfsck_layout *src)
  99 {
 100         int i;
 101
 102         des->ll_magic = cpu_to_le32(src->ll_magic);
 103         des->ll_status = cpu_to_le32(src->ll_status);
 104         des->ll_flags = cpu_to_le32(src->ll_flags);
 105         des->ll_success_count = cpu_to_le32(src->ll_success_count);
 106         des->ll_run_time_phase1 = cpu_to_le32(src->ll_run_time_phase1);
 107         des->ll_run_time_phase2 = cpu_to_le32(src->ll_run_time_phase2);
 108         des->ll_time_last_complete = cpu_to_le64(src->ll_time_last_complete);
 109         des->ll_time_latest_start = cpu_to_le64(src->ll_time_latest_start);
 110         des->ll_time_last_checkpoint =
 111                                 cpu_to_le64(src->ll_time_last_checkpoint);
 112         des->ll_pos_latest_start = cpu_to_le64(src->ll_pos_latest_start);
 113         des->ll_pos_last_checkpoint = cpu_to_le64(src->ll_pos_last_checkpoint);
 114         des->ll_pos_first_inconsistent =
 115                         cpu_to_le64(src->ll_pos_first_inconsistent);
 116         des->ll_objs_checked_phase1 = cpu_to_le64(src->ll_objs_checked_phase1);
 117         des->ll_objs_failed_phase1 = cpu_to_le64(src->ll_objs_failed_phase1);
 118         des->ll_objs_checked_phase2 = cpu_to_le64(src->ll_objs_checked_phase2);
 119         des->ll_objs_failed_phase2 = cpu_to_le64(src->ll_objs_failed_phase2);
 120         for (i = 0; i < LLIT_MAX; i++)
 121                 des->ll_objs_repaired[i] =
 122                                 cpu_to_le64(src->ll_objs_repaired[i]);
 123         des->ll_objs_skipped = cpu_to_le64(src->ll_objs_skipped);
 124 }
 125
 126 /**
 127  * \retval +ve: the lfsck_layout is broken, the caller should reset it.
 128  * \retval 0: succeed.
 129  * \retval -ve: failed cases.
 130  */
 131 static int lfsck_layout_load(const struct lu_env *env,
 132                              struct lfsck_component *com)
 133 {
 134         struct lfsck_layout             *lo     = com->lc_file_ram;
 135         const struct dt_body_operations *dbo    = com->lc_obj->do_body_ops;
 136         ssize_t                          size   = com->lc_file_size;
 137         loff_t                           pos    = 0;
 138         int                              rc;
 139
 140         rc = dbo->dbo_read(env, com->lc_obj,
 141                            lfsck_buf_get(env, com->lc_file_disk, size), &pos,
 142                            BYPASS_CAPA);
 143         if (rc == 0) {
 144                 return -ENOENT;
 145         } else if (rc < 0) {
 146                 CWARN("%s: failed to load lfsck_layout: rc = %d\n",
 147                       lfsck_lfsck2name(com->lc_lfsck), rc);
 148                 return rc;
 149         } else if (rc != size) {
 150                 CWARN("%s: crashed lfsck_layout, to be reset: rc = %d\n",
 151                       lfsck_lfsck2name(com->lc_lfsck), rc);
 152                 return 1;
 153         }
 154
 155         lfsck_layout_le_to_cpu(lo, com->lc_file_disk);
 156         if (lo->ll_magic != LFSCK_LAYOUT_MAGIC) {
 157                 CWARN("%s: invalid lfsck_layout magic %#x != %#x, "
 158                       "to be reset\n", lfsck_lfsck2name(com->lc_lfsck),
 159                       lo->ll_magic, LFSCK_LAYOUT_MAGIC);
 160                 return 1;
 161         }
 162
 163         return 0;
 164 }
 165
 166 static int lfsck_layout_store(const struct lu_env *env,
 167                               struct lfsck_component *com)
 168 {
 169         struct dt_object         *obj           = com->lc_obj;
 170         struct lfsck_instance    *lfsck         = com->lc_lfsck;
 171         struct lfsck_layout      *lo            = com->lc_file_disk;
 172         struct thandle           *handle;
 173         ssize_t                   size          = com->lc_file_size;
 174         loff_t                    pos           = 0;
 175         int                       rc;
 176         ENTRY;
 177
 178         lfsck_layout_cpu_to_le(lo, com->lc_file_ram);
 179         handle = dt_trans_create(env, lfsck->li_bottom);
 180         if (IS_ERR(handle)) {
 181                 rc = PTR_ERR(handle);
 182                 CERROR("%s: fail to create trans for storing lfsck_layout: "
 183                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 184                 RETURN(rc);
 185         }
 186
 187         rc = dt_declare_record_write(env, obj, size, pos, handle);
 188         if (rc != 0) {
 189                 CERROR("%s: fail to declare trans for storing lfsck_layout(1): "
 190                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 191                 GOTO(out, rc);
 192         }
 193
 194         rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
 195         if (rc != 0) {
 196                 CERROR("%s: fail to start trans for storing lfsck_layout: "
 197                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 198                 GOTO(out, rc);
 199         }
 200
 201         rc = dt_record_write(env, obj, lfsck_buf_get(env, lo, size), &pos,
 202                              handle);
 203         if (rc != 0)
 204                 CERROR("%s: fail to store lfsck_layout(1): size = %d, "
 205                        "rc = %d\n", lfsck_lfsck2name(lfsck), (int)size, rc);
 206
 207         GOTO(out, rc);
 208
 209 out:
 210         dt_trans_stop(env, lfsck->li_bottom, handle);
 211
 212         return rc;
 213 }
 214
 215 static int lfsck_layout_init(const struct lu_env *env,
 216                              struct lfsck_component *com)
 217 {
 218         struct lfsck_layout *lo = com->lc_file_ram;
 219         int rc;
 220
 221         memset(lo, 0, com->lc_file_size);
 222         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
 223         lo->ll_status = LS_INIT;
 224         down_write(&com->lc_sem);
 225         rc = lfsck_layout_store(env, com);
 226         up_write(&com->lc_sem);
 227
 228         return rc;
 229 }
 230
 231 static int fid_is_for_ostobj(const struct lu_env *env, struct dt_device *dt,
 232                              struct dt_object *obj, const struct lu_fid *fid)
 233 {
 234         struct seq_server_site  *ss     = lu_site2seq(dt->dd_lu_dev.ld_site);
 235         struct lu_seq_range      range  = { 0 };
 236         struct lustre_mdt_attrs *lma;
 237         int                      rc;
 238
 239         fld_range_set_any(&range);
 240         rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), &range);
 241         if (rc == 0) {
 242                 if (fld_range_is_ost(&range))
 243                         return 1;
 244
 245                 return 0;
 246         }
 247
 248         lma = &lfsck_env_info(env)->lti_lma;
 249         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, lma, sizeof(*lma)),
 250                           XATTR_NAME_LMA, BYPASS_CAPA);
 251         if (rc == sizeof(*lma)) {
 252                 lustre_lma_swab(lma);
 253
 254                 /* Generally, the low layer OSD create handler or OI scrub
 255                  * will set the LMAC_FID_ON_OST for all external visible
 256                  * OST-objects. But to make the otable-based iteration to
 257                  * be independent from OI scrub in spite of it got failure
 258                  * or not, we check the LMAC_FID_ON_OST here to guarantee
 259                  * that the LFSCK will not repair something by wrong. */
 260                 return lma->lma_compat & LMAC_FID_ON_OST ? 1 : 0;
 261         }
 262
 263         rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_FID, BYPASS_CAPA);
 264
 265         return rc > 0;
 266 }
 267
 268 static struct lfsck_layout_seq *
 269 lfsck_layout_seq_lookup(struct lfsck_layout_slave_data *llsd, __u64 seq)
 270 {
 271         struct lfsck_layout_seq *lls;
 272
 273         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
 274                 if (lls->lls_seq == seq)
 275                         return lls;
 276
 277                 if (lls->lls_seq > seq)
 278                         return NULL;
 279         }
 280
 281         return NULL;
 282 }
 283
 284 static void
 285 lfsck_layout_seq_insert(struct lfsck_layout_slave_data *llsd,
 286                         struct lfsck_layout_seq *lls)
 287 {
 288         struct lfsck_layout_seq *tmp;
 289         struct list_head        *pos = &llsd->llsd_seq_list;
 290
 291         list_for_each_entry(tmp, &llsd->llsd_seq_list, lls_list) {
 292                 if (lls->lls_seq < tmp->lls_seq) {
 293                         pos = &tmp->lls_list;
 294                         break;
 295                 }
 296         }
 297         list_add_tail(&lls->lls_list, pos);
 298 }
 299
 300 static int
 301 lfsck_layout_lastid_create(const struct lu_env *env,
 302                            struct lfsck_instance *lfsck,
 303                            struct dt_object *obj)
 304 {
 305         struct lfsck_thread_info *info   = lfsck_env_info(env);
 306         struct lu_attr           *la     = &info->lti_la;
 307         struct dt_object_format  *dof    = &info->lti_dof;
 308         struct lfsck_bookmark    *bk     = &lfsck->li_bookmark_ram;
 309         struct dt_device         *dt     = lfsck->li_bottom;
 310         struct thandle           *th;
 311         __u64                     lastid = 0;
 312         loff_t                    pos    = 0;
 313         int                       rc;
 314         ENTRY;
 315
 316         CDEBUG(D_LFSCK, "To create LAST_ID for <seq> "LPX64"\n",
 317                fid_seq(lfsck_dto2fid(obj)));
 318
 319         if (bk->lb_param & LPF_DRYRUN)
 320                 return 0;
 321
 322         memset(la, 0, sizeof(*la));
 323         la->la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
 324         la->la_valid = LA_MODE | LA_UID | LA_GID;
 325         dof->dof_type = dt_mode_to_dft(S_IFREG);
 326
 327         th = dt_trans_create(env, dt);
 328         if (IS_ERR(th))
 329                 RETURN(rc = PTR_ERR(th));
 330
 331         rc = dt_declare_create(env, obj, la, NULL, dof, th);
 332         if (rc != 0)
 333                 GOTO(stop, rc);
 334
 335         rc = dt_declare_record_write(env, obj, sizeof(lastid), pos, th);
 336         if (rc != 0)
 337                 GOTO(stop, rc);
 338
 339         rc = dt_trans_start_local(env, dt, th);
 340         if (rc != 0)
 341                 GOTO(stop, rc);
 342
 343         dt_write_lock(env, obj, 0);
 344         if (likely(!dt_object_exists(obj))) {
 345                 rc = dt_create(env, obj, la, NULL, dof, th);
 346                 if (rc == 0)
 347                         rc = dt_record_write(env, obj,
 348                                 lfsck_buf_get(env, &lastid, sizeof(lastid)),
 349                                 &pos, th);
 350         }
 351         dt_write_unlock(env, obj);
 352
 353         GOTO(stop, rc);
 354
 355 stop:
 356         dt_trans_stop(env, dt, th);
 357
 358         return rc;
 359 }
 360
 361 static int
 362 lfsck_layout_lastid_reload(const struct lu_env *env,
 363                            struct lfsck_component *com,
 364                            struct lfsck_layout_seq *lls)
 365 {
 366         __u64   lastid;
 367         loff_t  pos     = 0;
 368         int     rc;
 369
 370         dt_read_lock(env, lls->lls_lastid_obj, 0);
 371         rc = dt_record_read(env, lls->lls_lastid_obj,
 372                             lfsck_buf_get(env, &lastid, sizeof(lastid)), &pos);
 373         dt_read_unlock(env, lls->lls_lastid_obj);
 374         if (unlikely(rc != 0))
 375                 return rc;
 376
 377         lastid = le64_to_cpu(lastid);
 378         if (lastid < lls->lls_lastid_known) {
 379                 struct lfsck_instance   *lfsck  = com->lc_lfsck;
 380                 struct lfsck_layout     *lo     = com->lc_file_ram;
 381
 382                 lls->lls_lastid = lls->lls_lastid_known;
 383                 lls->lls_dirty = 1;
 384                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
 385                         LASSERT(lfsck->li_out_notify != NULL);
 386
 387                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 388                                              LE_LASTID_REBUILDING);
 389                         lo->ll_flags |= LF_CRASHED_LASTID;
 390                 }
 391         } else if (lastid >= lls->lls_lastid) {
 392                 lls->lls_lastid = lastid;
 393                 lls->lls_dirty = 0;
 394         }
 395
 396         return 0;
 397 }
 398
 399 static int
 400 lfsck_layout_lastid_store(const struct lu_env *env,
 401                           struct lfsck_component *com)
 402 {
 403         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 404         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
 405         struct dt_device                *dt     = lfsck->li_bottom;
 406         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
 407         struct lfsck_layout_seq         *lls;
 408         struct thandle                  *th;
 409         __u64                            lastid;
 410         int                              rc     = 0;
 411         int                              rc1    = 0;
 412
 413         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
 414                 loff_t pos = 0;
 415
 416                 /* XXX: Add the code back if we really found related
 417                  *      inconsistent cases in the future. */
 418 #if 0
 419                 if (!lls->lls_dirty) {
 420                         /* In OFD, before the pre-creation, the LAST_ID
 421                          * file will be updated firstly, which may hide
 422                          * some potential crashed cases. For example:
 423                          *
 424                          * The old obj1's ID is higher than old LAST_ID
 425                          * but lower than the new LAST_ID, but the LFSCK
 426                          * have not touch the obj1 until the OFD updated
 427                          * the LAST_ID. So the LFSCK does not regard it
 428                          * as crashed case. But when OFD does not create
 429                          * successfully, it will set the LAST_ID as the
 430                          * real created objects' ID, then LFSCK needs to
 431                          * found related inconsistency. */
 432                         rc = lfsck_layout_lastid_reload(env, com, lls);
 433                         if (likely(!lls->lls_dirty))
 434                                 continue;
 435                 }
 436 #endif
 437
 438                 CDEBUG(D_LFSCK, "To sync the LAST_ID for <seq> "LPX64
 439                        " as <oid> "LPU64"\n", lls->lls_seq, lls->lls_lastid);
 440
 441                 if (bk->lb_param & LPF_DRYRUN) {
 442                         lls->lls_dirty = 0;
 443                         continue;
 444                 }
 445
 446                 th = dt_trans_create(env, dt);
 447                 if (IS_ERR(th)) {
 448                         rc1 = PTR_ERR(th);
 449                         CERROR("%s: (1) failed to store "LPX64": rc = %d\n",
 450                                lfsck_lfsck2name(com->lc_lfsck),
 451                                lls->lls_seq, rc1);
 452                         continue;
 453                 }
 454
 455                 rc = dt_declare_record_write(env, lls->lls_lastid_obj,
 456                                              sizeof(lastid), pos, th);
 457                 if (rc != 0)
 458                         goto stop;
 459
 460                 rc = dt_trans_start_local(env, dt, th);
 461                 if (rc != 0)
 462                         goto stop;
 463
 464                 lastid = cpu_to_le64(lls->lls_lastid);
 465                 dt_write_lock(env, lls->lls_lastid_obj, 0);
 466                 rc = dt_record_write(env, lls->lls_lastid_obj,
 467                                      lfsck_buf_get(env, &lastid,
 468                                      sizeof(lastid)), &pos, th);
 469                 dt_write_unlock(env, lls->lls_lastid_obj);
 470                 if (rc == 0)
 471                         lls->lls_dirty = 0;
 472
 473 stop:
 474                 dt_trans_stop(env, dt, th);
 475                 if (rc != 0) {
 476                         rc1 = rc;
 477                         CERROR("%s: (2) failed to store "LPX64": rc = %d\n",
 478                                lfsck_lfsck2name(com->lc_lfsck),
 479                                lls->lls_seq, rc1);
 480                 }
 481         }
 482
 483         return rc1;
 484 }
 485
 486 static int
 487 lfsck_layout_lastid_load(const struct lu_env *env,
 488                          struct lfsck_component *com,
 489                          struct lfsck_layout_seq *lls)
 490 {
 491         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 492         struct lfsck_layout     *lo     = com->lc_file_ram;
 493         struct lu_fid           *fid    = &lfsck_env_info(env)->lti_fid;
 494         struct dt_object        *obj;
 495         loff_t                   pos    = 0;
 496         int                      rc;
 497         ENTRY;
 498
 499         lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck->li_bottom));
 500         obj = dt_locate(env, lfsck->li_bottom, fid);
 501         if (IS_ERR(obj))
 502                 RETURN(PTR_ERR(obj));
 503
 504         /* LAST_ID crashed, to be rebuilt */
 505         if (!dt_object_exists(obj)) {
 506                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
 507                         LASSERT(lfsck->li_out_notify != NULL);
 508
 509                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 510                                              LE_LASTID_REBUILDING);
 511                         lo->ll_flags |= LF_CRASHED_LASTID;
 512
 513                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY4) &&
 514                             cfs_fail_val > 0) {
 515                                 struct l_wait_info lwi = LWI_TIMEOUT(
 516                                                 cfs_time_seconds(cfs_fail_val),
 517                                                 NULL, NULL);
 518
 519                                 up_write(&com->lc_sem);
 520                                 l_wait_event(lfsck->li_thread.t_ctl_waitq,
 521                                              !thread_is_running(&lfsck->li_thread),
 522                                              &lwi);
 523                                 down_write(&com->lc_sem);
 524                         }
 525                 }
 526
 527                 rc = lfsck_layout_lastid_create(env, lfsck, obj);
 528         } else {
 529                 dt_read_lock(env, obj, 0);
 530                 rc = dt_read(env, obj,
 531                         lfsck_buf_get(env, &lls->lls_lastid, sizeof(__u64)),
 532                         &pos);
 533                 dt_read_unlock(env, obj);
 534                 if (rc != 0 && rc != sizeof(__u64))
 535                         GOTO(out, rc = (rc > 0 ? -EFAULT : rc));
 536
 537                 if (rc == 0 && !(lo->ll_flags & LF_CRASHED_LASTID)) {
 538                         LASSERT(lfsck->li_out_notify != NULL);
 539
 540                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 541                                              LE_LASTID_REBUILDING);
 542                         lo->ll_flags |= LF_CRASHED_LASTID;
 543                 }
 544
 545                 lls->lls_lastid = le64_to_cpu(lls->lls_lastid);
 546                 rc = 0;
 547         }
 548
 549         GOTO(out, rc);
 550
 551 out:
 552         if (rc != 0)
 553                 lfsck_object_put(env, obj);
 554         else
 555                 lls->lls_lastid_obj = obj;
 556
 557         return rc;
 558 }
 559
 560 /* layout APIs */
 561 /* XXX: Some to be implemented in other patch(es). */
 562
 563 static int lfsck_layout_reset(const struct lu_env *env,
 564                               struct lfsck_component *com, bool init)
 565 {
 566         struct lfsck_layout     *lo    = com->lc_file_ram;
 567         int                      rc;
 568
 569         down_write(&com->lc_sem);
 570         if (init) {
 571                 memset(lo, 0, com->lc_file_size);
 572         } else {
 573                 __u32 count = lo->ll_success_count;
 574                 __u64 last_time = lo->ll_time_last_complete;
 575
 576                 memset(lo, 0, com->lc_file_size);
 577                 lo->ll_success_count = count;
 578                 lo->ll_time_last_complete = last_time;
 579         }
 580
 581         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
 582         lo->ll_status = LS_INIT;
 583
 584         rc = lfsck_layout_store(env, com);
 585         up_write(&com->lc_sem);
 586
 587         return rc;
 588 }
 589
 590 static void lfsck_layout_fail(const struct lu_env *env,
 591                               struct lfsck_component *com, bool new_checked)
 592 {
 593         struct lfsck_layout *lo = com->lc_file_ram;
 594
 595         down_write(&com->lc_sem);
 596         if (new_checked)
 597                 com->lc_new_checked++;
 598         lo->ll_objs_failed_phase1++;
 599         if (lo->ll_pos_first_inconsistent == 0) {
 600                 struct lfsck_instance *lfsck = com->lc_lfsck;
 601
 602                 lo->ll_pos_first_inconsistent =
 603                         lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
 604                                                         lfsck->li_di_oit);
 605         }
 606         up_write(&com->lc_sem);
 607 }
 608
 609 static int lfsck_layout_checkpoint(const struct lu_env *env,
 610                                    struct lfsck_component *com, bool init)
 611 {
 612         struct lfsck_instance   *lfsck = com->lc_lfsck;
 613         struct lfsck_layout     *lo    = com->lc_file_ram;
 614         int                      rc;
 615
 616         if (com->lc_new_checked == 0 && !init)
 617                 return 0;
 618
 619         down_write(&com->lc_sem);
 620
 621         if (init) {
 622                 lo->ll_pos_latest_start = lfsck->li_pos_current.lp_oit_cookie;
 623         } else {
 624                 lo->ll_pos_last_checkpoint =
 625                                         lfsck->li_pos_current.lp_oit_cookie;
 626                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
 627                                 HALF_SEC - lfsck->li_time_last_checkpoint);
 628                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
 629                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
 630                 com->lc_new_checked = 0;
 631         }
 632
 633         rc = lfsck_layout_store(env, com);
 634
 635         up_write(&com->lc_sem);
 636
 637         return rc;
 638 }
 639
 640 static int lfsck_layout_master_prep(const struct lu_env *env,
 641                                     struct lfsck_component *com)
 642 {
 643         return 0;
 644 }
 645
 646 static int lfsck_layout_slave_prep(const struct lu_env *env,
 647                                    struct lfsck_component *com)
 648 {
 649         struct lfsck_instance   *lfsck  = com->lc_lfsck;
 650         struct lfsck_layout     *lo     = com->lc_file_ram;
 651         struct lfsck_position   *pos    = &com->lc_pos_start;
 652
 653         /* XXX: For a new scanning, generate OST-objects
 654          *      bitmap for orphan detection. */
 655
 656         fid_zero(&pos->lp_dir_parent);
 657         pos->lp_dir_cookie = 0;
 658         if (lo->ll_status == LS_COMPLETED ||
 659             lo->ll_status == LS_PARTIAL) {
 660                 int rc;
 661
 662                 rc = lfsck_layout_reset(env, com, false);
 663                 if (rc != 0)
 664                         return rc;
 665         }
 666
 667         down_write(&com->lc_sem);
 668
 669         lo->ll_time_latest_start = cfs_time_current_sec();
 670
 671         spin_lock(&lfsck->li_lock);
 672         if (lo->ll_flags & LF_SCANNED_ONCE) {
 673                 if (!lfsck->li_drop_dryrun ||
 674                     lo->ll_pos_first_inconsistent == 0) {
 675                         lo->ll_status = LS_SCANNING_PHASE2;
 676                         list_del_init(&com->lc_link);
 677                         list_add_tail(&com->lc_link,
 678                                       &lfsck->li_list_double_scan);
 679                         pos->lp_oit_cookie = 0;
 680                 } else {
 681                         int i;
 682
 683                         lo->ll_status = LS_SCANNING_PHASE1;
 684                         lo->ll_run_time_phase1 = 0;
 685                         lo->ll_run_time_phase2 = 0;
 686                         lo->ll_objs_checked_phase1 = 0;
 687                         lo->ll_objs_checked_phase2 = 0;
 688                         lo->ll_objs_failed_phase1 = 0;
 689                         lo->ll_objs_failed_phase2 = 0;
 690                         for (i = 0; i < LLIT_MAX; i++)
 691                                 lo->ll_objs_repaired[i] = 0;
 692
 693                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
 694                 }
 695         } else {
 696                 lo->ll_status = LS_SCANNING_PHASE1;
 697                 if (!lfsck->li_drop_dryrun ||
 698                     lo->ll_pos_first_inconsistent == 0)
 699                         pos->lp_oit_cookie = lo->ll_pos_last_checkpoint + 1;
 700                 else
 701                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
 702         }
 703         spin_unlock(&lfsck->li_lock);
 704
 705         up_write(&com->lc_sem);
 706
 707         return 0;
 708 }
 709
 710 static int lfsck_layout_master_exec_oit(const struct lu_env *env,
 711                                         struct lfsck_component *com,
 712                                         struct dt_object *obj)
 713 {
 714         return 0;
 715 }
 716
 717 static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
 718                                        struct lfsck_component *com,
 719                                        struct dt_object *obj)
 720 {
 721         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 722         struct lfsck_layout             *lo     = com->lc_file_ram;
 723         const struct lu_fid             *fid    = lfsck_dto2fid(obj);
 724         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
 725         struct lfsck_layout_seq         *lls;
 726         __u64                            seq;
 727         __u64                            oid;
 728         int                              rc;
 729         ENTRY;
 730
 731         /* XXX: Update OST-objects bitmap for orphan detection. */
 732
 733         LASSERT(llsd != NULL);
 734
 735         down_write(&com->lc_sem);
 736         if (fid_is_idif(fid))
 737                 seq = 0;
 738         else if (!fid_is_norm(fid) ||
 739                  !fid_is_for_ostobj(env, lfsck->li_next, obj, fid))
 740                 GOTO(unlock, rc = 0);
 741         else
 742                 seq = fid_seq(fid);
 743         com->lc_new_checked++;
 744
 745         lls = lfsck_layout_seq_lookup(llsd, seq);
 746         if (lls == NULL) {
 747                 OBD_ALLOC_PTR(lls);
 748                 if (unlikely(lls == NULL))
 749                         GOTO(unlock, rc = -ENOMEM);
 750
 751                 INIT_LIST_HEAD(&lls->lls_list);
 752                 lls->lls_seq = seq;
 753                 rc = lfsck_layout_lastid_load(env, com, lls);
 754                 if (rc != 0) {
 755                         lo->ll_objs_failed_phase1++;
 756                         OBD_FREE_PTR(lls);
 757                         GOTO(unlock, rc);
 758                 }
 759
 760                 lfsck_layout_seq_insert(llsd, lls);
 761         }
 762
 763         if (unlikely(fid_is_last_id(fid)))
 764                 GOTO(unlock, rc = 0);
 765
 766         oid = fid_oid(fid);
 767         if (oid > lls->lls_lastid_known)
 768                 lls->lls_lastid_known = oid;
 769
 770         if (oid > lls->lls_lastid) {
 771                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
 772                         /* OFD may create new objects during LFSCK scanning. */
 773                         rc = lfsck_layout_lastid_reload(env, com, lls);
 774                         if (unlikely(rc != 0))
 775                                 CWARN("%s: failed to reload LAST_ID for "LPX64
 776                                       ": rc = %d\n",
 777                                       lfsck_lfsck2name(com->lc_lfsck),
 778                                       lls->lls_seq, rc);
 779                         if (oid <= lls->lls_lastid)
 780                                 GOTO(unlock, rc = 0);
 781
 782                         LASSERT(lfsck->li_out_notify != NULL);
 783
 784                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 785                                              LE_LASTID_REBUILDING);
 786                         lo->ll_flags |= LF_CRASHED_LASTID;
 787                 }
 788
 789                 lls->lls_lastid = oid;
 790                 lls->lls_dirty = 1;
 791         }
 792
 793         GOTO(unlock, rc = 0);
 794
 795 unlock:
 796         up_write(&com->lc_sem);
 797
 798         return rc;
 799 }
 800
 801 static int lfsck_layout_exec_dir(const struct lu_env *env,
 802                                  struct lfsck_component *com,
 803                                  struct dt_object *obj,
 804                                  struct lu_dirent *ent)
 805 {
 806         return 0;
 807 }
 808
 809 static int lfsck_layout_master_post(const struct lu_env *env,
 810                                     struct lfsck_component *com,
 811                                     int result, bool init)
 812 {
 813         return 0;
 814 }
 815
 816 static int lfsck_layout_slave_post(const struct lu_env *env,
 817                                    struct lfsck_component *com,
 818                                    int result, bool init)
 819 {
 820         struct lfsck_instance   *lfsck = com->lc_lfsck;
 821         struct lfsck_layout     *lo    = com->lc_file_ram;
 822         int                      rc;
 823         bool                     done  = false;
 824
 825         rc = lfsck_layout_lastid_store(env, com);
 826         if (rc != 0)
 827                 result = rc;
 828
 829         LASSERT(lfsck->li_out_notify != NULL);
 830
 831         down_write(&com->lc_sem);
 832
 833         spin_lock(&lfsck->li_lock);
 834         if (!init)
 835                 lo->ll_pos_last_checkpoint =
 836                                         lfsck->li_pos_current.lp_oit_cookie;
 837         if (result > 0) {
 838                 lo->ll_status = LS_SCANNING_PHASE2;
 839                 lo->ll_flags |= LF_SCANNED_ONCE;
 840                 if (lo->ll_flags & LF_CRASHED_LASTID) {
 841                         done = true;
 842                         lo->ll_flags &= ~LF_CRASHED_LASTID;
 843                 }
 844                 lo->ll_flags &= ~LF_UPGRADE;
 845                 list_del_init(&com->lc_link);
 846                 list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
 847         } else if (result == 0) {
 848                 if (lfsck->li_paused) {
 849                         lo->ll_status = LS_PAUSED;
 850                 } else {
 851                         lo->ll_status = LS_STOPPED;
 852                         list_del_init(&com->lc_link);
 853                         list_add_tail(&com->lc_link, &lfsck->li_list_idle);
 854                 }
 855         } else {
 856                 lo->ll_status = LS_FAILED;
 857                 list_del_init(&com->lc_link);
 858                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
 859         }
 860         spin_unlock(&lfsck->li_lock);
 861
 862         if (done)
 863                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 864                                      LE_LASTID_REBUILT);
 865
 866         if (!init) {
 867                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
 868                                 HALF_SEC - lfsck->li_time_last_checkpoint);
 869                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
 870                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
 871                 com->lc_new_checked = 0;
 872         }
 873
 874         rc = lfsck_layout_store(env, com);
 875
 876         up_write(&com->lc_sem);
 877
 878         return rc;
 879 }
 880
 881 static int lfsck_layout_dump(const struct lu_env *env,
 882                              struct lfsck_component *com, char *buf, int len)
 883 {
 884         struct lfsck_instance   *lfsck = com->lc_lfsck;
 885         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
 886         struct lfsck_layout     *lo    = com->lc_file_ram;
 887         int                      save  = len;
 888         int                      ret   = -ENOSPC;
 889         int                      rc;
 890
 891         down_read(&com->lc_sem);
 892         rc = snprintf(buf, len,
 893                       "name: lfsck_layout\n"
 894                       "magic: %#x\n"
 895                       "version: %d\n"
 896                       "status: %s\n",
 897                       lo->ll_magic,
 898                       bk->lb_version,
 899                       lfsck_status2names(lo->ll_status));
 900         if (rc <= 0)
 901                 goto out;
 902
 903         buf += rc;
 904         len -= rc;
 905         rc = lfsck_bits_dump(&buf, &len, lo->ll_flags, lfsck_flags_names,
 906                              "flags");
 907         if (rc < 0)
 908                 goto out;
 909
 910         rc = lfsck_bits_dump(&buf, &len, bk->lb_param, lfsck_param_names,
 911                              "param");
 912         if (rc < 0)
 913                 goto out;
 914
 915         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_complete,
 916                              "time_since_last_completed");
 917         if (rc < 0)
 918                 goto out;
 919
 920         rc = lfsck_time_dump(&buf, &len, lo->ll_time_latest_start,
 921                              "time_since_latest_start");
 922         if (rc < 0)
 923                 goto out;
 924
 925         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_checkpoint,
 926                              "time_since_last_checkpoint");
 927         if (rc < 0)
 928                 goto out;
 929
 930         rc = snprintf(buf, len,
 931                       "latest_start_position: "LPU64"\n"
 932                       "last_checkpoint_position: "LPU64"\n"
 933                       "first_failure_position: "LPU64"\n",
 934                       lo->ll_pos_latest_start,
 935                       lo->ll_pos_last_checkpoint,
 936                       lo->ll_pos_first_inconsistent);
 937         if (rc <= 0)
 938                 goto out;
 939
 940         buf += rc;
 941         len -= rc;
 942
 943         rc = snprintf(buf, len,
 944                       "success_count: %u\n"
 945                       "repaired_dangling: "LPU64"\n"
 946                       "repaired_unmatched_pair: "LPU64"\n"
 947                       "repaired_multiple_referenced: "LPU64"\n"
 948                       "repaired_orphan: "LPU64"\n"
 949                       "repaired_inconsistent_owner: "LPU64"\n"
 950                       "repaired_others: "LPU64"\n"
 951                       "skipped: "LPU64"\n"
 952                       "failed_phase1: "LPU64"\n"
 953                       "failed_phase2: "LPU64"\n",
 954                       lo->ll_success_count,
 955                       lo->ll_objs_repaired[LLIT_DANGLING - 1],
 956                       lo->ll_objs_repaired[LLIT_UNMATCHED_PAIR - 1],
 957                       lo->ll_objs_repaired[LLIT_MULTIPLE_REFERENCED - 1],
 958                       lo->ll_objs_repaired[LLIT_ORPHAN - 1],
 959                       lo->ll_objs_repaired[LLIT_INCONSISTENT_OWNER - 1],
 960                       lo->ll_objs_repaired[LLIT_OTHERS - 1],
 961                       lo->ll_objs_skipped,
 962                       lo->ll_objs_failed_phase1,
 963                       lo->ll_objs_failed_phase2);
 964         if (rc <= 0)
 965                 goto out;
 966
 967         buf += rc;
 968         len -= rc;
 969
 970         if (lo->ll_status == LS_SCANNING_PHASE1) {
 971                 __u64 pos;
 972                 const struct dt_it_ops *iops;
 973                 cfs_duration_t duration = cfs_time_current() -
 974                                           lfsck->li_time_last_checkpoint;
 975                 __u64 checked = lo->ll_objs_checked_phase1 + com->lc_new_checked;
 976                 __u64 speed = checked;
 977                 __u64 new_checked = com->lc_new_checked * HZ;
 978                 __u32 rtime = lo->ll_run_time_phase1 +
 979                               cfs_duration_sec(duration + HALF_SEC);
 980
 981                 if (duration != 0)
 982                         do_div(new_checked, duration);
 983                 if (rtime != 0)
 984                         do_div(speed, rtime);
 985                 rc = snprintf(buf, len,
 986                               "checked_phase1: "LPU64"\n"
 987                               "checked_phase2: "LPU64"\n"
 988                               "run_time_phase1: %u seconds\n"
 989                               "run_time_phase2: %u seconds\n"
 990                               "average_speed_phase1: "LPU64" items/sec\n"
 991                               "average_speed_phase2: N/A\n"
 992                               "real-time_speed_phase1: "LPU64" items/sec\n"
 993                               "real-time_speed_phase2: N/A\n",
 994                               checked,
 995                               lo->ll_objs_checked_phase2,
 996                               rtime,
 997                               lo->ll_run_time_phase2,
 998                               speed,
 999                               new_checked);
1000                 if (rc <= 0)
1001                         goto out;
1002
1003                 buf += rc;
1004                 len -= rc;
1005
1006                 LASSERT(lfsck->li_di_oit != NULL);
1007
1008                 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
1009
1010                 /* The low layer otable-based iteration position may NOT
1011                  * exactly match the layout-based directory traversal
1012                  * cookie. Generally, it is not a serious issue. But the
1013                  * caller should NOT make assumption on that. */
1014                 pos = iops->store(env, lfsck->li_di_oit);
1015                 if (!lfsck->li_current_oit_processed)
1016                         pos--;
1017                 rc = snprintf(buf, len, "current_position: "LPU64"\n", pos);
1018                 if (rc <= 0)
1019                         goto out;
1020
1021                 buf += rc;
1022                 len -= rc;
1023         } else {
1024                 /* XXX: LS_SCANNING_PHASE2 will be handled in the future. */
1025                 __u64 speed1 = lo->ll_objs_checked_phase1;
1026                 __u64 speed2 = lo->ll_objs_checked_phase2;
1027
1028                 if (lo->ll_run_time_phase1 != 0)
1029                         do_div(speed1, lo->ll_run_time_phase1);
1030                 if (lo->ll_run_time_phase2 != 0)
1031                         do_div(speed2, lo->ll_run_time_phase2);
1032                 rc = snprintf(buf, len,
1033                               "checked_phase1: "LPU64"\n"
1034                               "checked_phase2: "LPU64"\n"
1035                               "run_time_phase1: %u seconds\n"
1036                               "run_time_phase2: %u seconds\n"
1037                               "average_speed_phase1: "LPU64" items/sec\n"
1038                               "average_speed_phase2: "LPU64" objs/sec\n"
1039                               "real-time_speed_phase1: N/A\n"
1040                               "real-time_speed_phase2: N/A\n"
1041                               "current_position: N/A\n",
1042                               lo->ll_objs_checked_phase1,
1043                               lo->ll_objs_checked_phase2,
1044                               lo->ll_run_time_phase1,
1045                               lo->ll_run_time_phase2,
1046                               speed1,
1047                               speed2);
1048                 if (rc <= 0)
1049                         goto out;
1050
1051                 buf += rc;
1052                 len -= rc;
1053         }
1054         ret = save - len;
1055
1056 out:
1057         up_read(&com->lc_sem);
1058
1059         return ret;
1060 }
1061
1062 static int lfsck_layout_master_double_scan(const struct lu_env *env,
1063                                            struct lfsck_component *com)
1064 {
1065         return 0;
1066 }
1067
1068 static int lfsck_layout_slave_double_scan(const struct lu_env *env,
1069                                           struct lfsck_component *com)
1070 {
1071         struct lfsck_instance   *lfsck = com->lc_lfsck;
1072         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
1073         struct lfsck_layout     *lo    = com->lc_file_ram;
1074         int                      rc    = 1;
1075
1076         down_write(&com->lc_sem);
1077
1078         lo->ll_run_time_phase2 += cfs_duration_sec(cfs_time_current() +
1079                                 HALF_SEC - lfsck->li_time_last_checkpoint);
1080         lo->ll_time_last_checkpoint = cfs_time_current_sec();
1081         lo->ll_objs_checked_phase2 += com->lc_new_checked;
1082
1083         com->lc_new_checked = 0;
1084         com->lc_new_scanned = 0;
1085         com->lc_time_last_checkpoint = cfs_time_current();
1086         com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
1087                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
1088
1089         if (rc > 0) {
1090                 com->lc_journal = 0;
1091                 if (lo->ll_flags & LF_INCOMPLETE)
1092                         lo->ll_status = LS_PARTIAL;
1093                 else
1094                         lo->ll_status = LS_COMPLETED;
1095                 if (!(bk->lb_param & LPF_DRYRUN))
1096                         lo->ll_flags &= ~(LF_SCANNED_ONCE | LF_INCONSISTENT);
1097                 lo->ll_time_last_complete = lo->ll_time_last_checkpoint;
1098                 lo->ll_success_count++;
1099         } else if (rc == 0) {
1100                 if (lfsck->li_paused)
1101                         lo->ll_status = LS_PAUSED;
1102                 else
1103                         lo->ll_status = LS_STOPPED;
1104         } else {
1105                 lo->ll_status = LS_FAILED;
1106         }
1107
1108         if (lo->ll_status != LS_PAUSED) {
1109                 spin_lock(&lfsck->li_lock);
1110                 list_del_init(&com->lc_link);
1111                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1112                 spin_unlock(&lfsck->li_lock);
1113         }
1114
1115         rc = lfsck_layout_store(env, com);
1116
1117         up_write(&com->lc_sem);
1118
1119         return rc;
1120 }
1121
1122 static void lfsck_layout_master_data_release(const struct lu_env *env,
1123                                              struct lfsck_component *com)
1124 {
1125 }
1126
1127 static void lfsck_layout_slave_data_release(const struct lu_env *env,
1128                                             struct lfsck_component *com)
1129 {
1130         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
1131         struct lfsck_layout_seq         *lls;
1132         struct lfsck_layout_seq         *next;
1133
1134         LASSERT(llsd != NULL);
1135
1136         com->lc_data = NULL;
1137
1138         list_for_each_entry_safe(lls, next, &llsd->llsd_seq_list,
1139                                      lls_list) {
1140                 list_del_init(&lls->lls_list);
1141                 lfsck_object_put(env, lls->lls_lastid_obj);
1142                 OBD_FREE_PTR(lls);
1143         }
1144
1145         OBD_FREE_PTR(llsd);
1146 }
1147
1148 static struct lfsck_operations lfsck_layout_master_ops = {
1149         .lfsck_reset            = lfsck_layout_reset,
1150         .lfsck_fail             = lfsck_layout_fail,
1151         .lfsck_checkpoint       = lfsck_layout_checkpoint,
1152         .lfsck_prep             = lfsck_layout_master_prep,
1153         .lfsck_exec_oit         = lfsck_layout_master_exec_oit,
1154         .lfsck_exec_dir         = lfsck_layout_exec_dir,
1155         .lfsck_post             = lfsck_layout_master_post,
1156         .lfsck_dump             = lfsck_layout_dump,
1157         .lfsck_double_scan      = lfsck_layout_master_double_scan,
1158         .lfsck_data_release     = lfsck_layout_master_data_release,
1159 };
1160
1161 static struct lfsck_operations lfsck_layout_slave_ops = {
1162         .lfsck_reset            = lfsck_layout_reset,
1163         .lfsck_fail             = lfsck_layout_fail,
1164         .lfsck_checkpoint       = lfsck_layout_checkpoint,
1165         .lfsck_prep             = lfsck_layout_slave_prep,
1166         .lfsck_exec_oit         = lfsck_layout_slave_exec_oit,
1167         .lfsck_exec_dir         = lfsck_layout_exec_dir,
1168         .lfsck_post             = lfsck_layout_slave_post,
1169         .lfsck_dump             = lfsck_layout_dump,
1170         .lfsck_double_scan      = lfsck_layout_slave_double_scan,
1171         .lfsck_data_release     = lfsck_layout_slave_data_release,
1172 };
1173
1174 int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
1175 {
1176         struct lfsck_component  *com;
1177         struct lfsck_layout     *lo;
1178         struct dt_object        *root = NULL;
1179         struct dt_object        *obj;
1180         int                      rc;
1181         ENTRY;
1182
1183         OBD_ALLOC_PTR(com);
1184         if (com == NULL)
1185                 RETURN(-ENOMEM);
1186
1187         INIT_LIST_HEAD(&com->lc_link);
1188         INIT_LIST_HEAD(&com->lc_link_dir);
1189         init_rwsem(&com->lc_sem);
1190         atomic_set(&com->lc_ref, 1);
1191         com->lc_lfsck = lfsck;
1192         com->lc_type = LT_LAYOUT;
1193         if (lfsck->li_master) {
1194                 com->lc_ops = &lfsck_layout_master_ops;
1195         } else {
1196                 struct lfsck_layout_slave_data *llsd;
1197
1198                 com->lc_ops = &lfsck_layout_slave_ops;
1199                 OBD_ALLOC_PTR(llsd);
1200                 if (llsd == NULL)
1201                         GOTO(out, rc = -ENOMEM);
1202
1203                 INIT_LIST_HEAD(&llsd->llsd_seq_list);
1204                 com->lc_data = llsd;
1205         }
1206         com->lc_file_size = sizeof(*lo);
1207         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
1208         if (com->lc_file_ram == NULL)
1209                 GOTO(out, rc = -ENOMEM);
1210
1211         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
1212         if (com->lc_file_disk == NULL)
1213                 GOTO(out, rc = -ENOMEM);
1214
1215         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
1216         if (IS_ERR(root))
1217                 GOTO(out, rc = PTR_ERR(root));
1218
1219         if (unlikely(!dt_try_as_dir(env, root)))
1220                 GOTO(out, rc = -ENOTDIR);
1221
1222         obj = local_file_find_or_create(env, lfsck->li_los, root,
1223                                         lfsck_layout_name,
1224                                         S_IFREG | S_IRUGO | S_IWUSR);
1225         if (IS_ERR(obj))
1226                 GOTO(out, rc = PTR_ERR(obj));
1227
1228         com->lc_obj = obj;
1229         rc = lfsck_layout_load(env, com);
1230         if (rc > 0)
1231                 rc = lfsck_layout_reset(env, com, true);
1232         else if (rc == -ENOENT)
1233                 rc = lfsck_layout_init(env, com);
1234
1235         if (rc != 0)
1236                 GOTO(out, rc);
1237
1238         lo = com->lc_file_ram;
1239         switch (lo->ll_status) {
1240         case LS_INIT:
1241         case LS_COMPLETED:
1242         case LS_FAILED:
1243         case LS_STOPPED:
1244         case LS_PARTIAL:
1245                 spin_lock(&lfsck->li_lock);
1246                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1247                 spin_unlock(&lfsck->li_lock);
1248                 break;
1249         default:
1250                 CERROR("%s: unknown lfsck_layout status: rc = %u\n",
1251                        lfsck_lfsck2name(lfsck), lo->ll_status);
1252                 /* fall through */
1253         case LS_SCANNING_PHASE1:
1254         case LS_SCANNING_PHASE2:
1255                 /* No need to store the status to disk right now.
1256                  * If the system crashed before the status stored,
1257                  * it will be loaded back when next time. */
1258                 lo->ll_status = LS_CRASHED;
1259                 lo->ll_flags |= LF_INCOMPLETE;
1260                 /* fall through */
1261         case LS_PAUSED:
1262         case LS_CRASHED:
1263                 spin_lock(&lfsck->li_lock);
1264                 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
1265                 spin_unlock(&lfsck->li_lock);
1266                 break;
1267         }
1268
1269         if (lo->ll_flags & LF_CRASHED_LASTID) {
1270                 LASSERT(lfsck->li_out_notify != NULL);
1271
1272                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1273                                      LE_LASTID_REBUILDING);
1274         }
1275
1276         GOTO(out, rc = 0);
1277
1278 out:
1279         if (root != NULL && !IS_ERR(root))
1280                 lu_object_put(env, &root->do_lu);
1281
1282         if (rc != 0)
1283                 lfsck_component_cleanup(env, com);
1284
1285         return rc;
1286 }