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 #include <linux/rbtree.h>
  38
  39 #include <lustre/lustre_idl.h>
  40 #include <lu_object.h>
  41 #include <dt_object.h>
  42 #include <lustre_linkea.h>
  43 #include <lustre_fid.h>
  44 #include <lustre_lib.h>
  45 #include <lustre_net.h>
  46 #include <lustre/lustre_user.h>
  47 #include <md_object.h>
  48 #include <obd_class.h>
  49
  50 #include "lfsck_internal.h"
  51
  52 #define LFSCK_LAYOUT_MAGIC              0xB173AE14
  53
  54 static const char lfsck_layout_name[] = "lfsck_layout";
  55
  56 struct lfsck_layout_seq {
  57         struct list_head         lls_list;
  58         __u64                    lls_seq;
  59         __u64                    lls_lastid;
  60         __u64                    lls_lastid_known;
  61         struct dt_object        *lls_lastid_obj;
  62         unsigned int             lls_dirty:1;
  63 };
  64
  65 struct lfsck_layout_slave_target {
  66         /* link into lfsck_layout_slave_data::llsd_master_list. */
  67         struct list_head        llst_list;
  68         /* The position for next record in the rbtree for iteration. */
  69         struct lu_fid           llst_fid;
  70         /* Dummy hash for iteration against the rbtree. */
  71         __u64                   llst_hash;
  72         __u64                   llst_gen;
  73         atomic_t                llst_ref;
  74         __u32                   llst_index;
  75 };
  76
  77 struct lfsck_layout_slave_data {
  78         /* list for lfsck_layout_seq */
  79         struct list_head         llsd_seq_list;
  80
  81         /* list for the masters involve layout verification. */
  82         struct list_head         llsd_master_list;
  83         spinlock_t               llsd_lock;
  84         __u64                    llsd_touch_gen;
  85         struct dt_object        *llsd_rb_obj;
  86         struct rb_root           llsd_rb_root;
  87         rwlock_t                 llsd_rb_lock;
  88         unsigned int             llsd_rbtree_valid:1;
  89 };
  90
  91 struct lfsck_layout_object {
  92         struct dt_object        *llo_obj;
  93         struct lu_attr           llo_attr;
  94         atomic_t                 llo_ref;
  95         __u16                    llo_gen;
  96 };
  97
  98 struct lfsck_layout_req {
  99         struct list_head                 llr_list;
 100         struct lfsck_layout_object      *llr_parent;
 101         struct dt_object                *llr_child;
 102         __u32                            llr_ost_idx;
 103         __u32                            llr_lov_idx; /* offset in LOV EA */
 104 };
 105
 106 struct lfsck_layout_master_data {
 107         spinlock_t              llmd_lock;
 108         struct list_head        llmd_req_list;
 109
 110         /* list for the ost targets involve layout verification. */
 111         struct list_head        llmd_ost_list;
 112
 113         /* list for the ost targets in phase1 scanning. */
 114         struct list_head        llmd_ost_phase1_list;
 115
 116         /* list for the ost targets in phase1 scanning. */
 117         struct list_head        llmd_ost_phase2_list;
 118
 119         /* list for the mdt targets involve layout verification. */
 120         struct list_head        llmd_mdt_list;
 121
 122         /* list for the mdt targets in phase1 scanning. */
 123         struct list_head        llmd_mdt_phase1_list;
 124
 125         /* list for the mdt targets in phase1 scanning. */
 126         struct list_head        llmd_mdt_phase2_list;
 127
 128         struct ptlrpc_thread    llmd_thread;
 129         __u32                   llmd_touch_gen;
 130         int                     llmd_prefetched;
 131         int                     llmd_assistant_status;
 132         int                     llmd_post_result;
 133         unsigned int            llmd_to_post:1,
 134                                 llmd_to_double_scan:1,
 135                                 llmd_in_double_scan:1,
 136                                 llmd_exit:1;
 137 };
 138
 139 struct lfsck_layout_slave_async_args {
 140         struct obd_export                *llsaa_exp;
 141         struct lfsck_component           *llsaa_com;
 142         struct lfsck_layout_slave_target *llsaa_llst;
 143 };
 144
 145 static struct lfsck_layout_object *
 146 lfsck_layout_object_init(const struct lu_env *env, struct dt_object *obj,
 147                          __u16 gen)
 148 {
 149         struct lfsck_layout_object *llo;
 150         int                         rc;
 151
 152         OBD_ALLOC_PTR(llo);
 153         if (llo == NULL)
 154                 return ERR_PTR(-ENOMEM);
 155
 156         rc = dt_attr_get(env, obj, &llo->llo_attr, BYPASS_CAPA);
 157         if (rc != 0) {
 158                 OBD_FREE_PTR(llo);
 159
 160                 return ERR_PTR(rc);
 161         }
 162
 163         lu_object_get(&obj->do_lu);
 164         llo->llo_obj = obj;
 165         /* The gen can be used to check whether some others have changed the
 166          * file layout after LFSCK pre-fetching but before real verification. */
 167         llo->llo_gen = gen;
 168         atomic_set(&llo->llo_ref, 1);
 169
 170         return llo;
 171 }
 172
 173 static inline void
 174 lfsck_layout_llst_put(struct lfsck_layout_slave_target *llst)
 175 {
 176         if (atomic_dec_and_test(&llst->llst_ref)) {
 177                 LASSERT(list_empty(&llst->llst_list));
 178
 179                 OBD_FREE_PTR(llst);
 180         }
 181 }
 182
 183 static inline int
 184 lfsck_layout_llst_add(struct lfsck_layout_slave_data *llsd, __u32 index)
 185 {
 186         struct lfsck_layout_slave_target *llst;
 187         struct lfsck_layout_slave_target *tmp;
 188         int                               rc   = 0;
 189
 190         OBD_ALLOC_PTR(llst);
 191         if (llst == NULL)
 192                 return -ENOMEM;
 193
 194         INIT_LIST_HEAD(&llst->llst_list);
 195         llst->llst_gen = 0;
 196         llst->llst_index = index;
 197         atomic_set(&llst->llst_ref, 1);
 198
 199         spin_lock(&llsd->llsd_lock);
 200         list_for_each_entry(tmp, &llsd->llsd_master_list, llst_list) {
 201                 if (tmp->llst_index == index) {
 202                         rc = -EALREADY;
 203                         break;
 204                 }
 205         }
 206         if (rc == 0)
 207                 list_add_tail(&llst->llst_list, &llsd->llsd_master_list);
 208         spin_unlock(&llsd->llsd_lock);
 209
 210         if (rc != 0)
 211                 OBD_FREE_PTR(llst);
 212
 213         return rc;
 214 }
 215
 216 static inline void
 217 lfsck_layout_llst_del(struct lfsck_layout_slave_data *llsd,
 218                       struct lfsck_layout_slave_target *llst)
 219 {
 220         bool del = false;
 221
 222         spin_lock(&llsd->llsd_lock);
 223         if (!list_empty(&llst->llst_list)) {
 224                 list_del_init(&llst->llst_list);
 225                 del = true;
 226         }
 227         spin_unlock(&llsd->llsd_lock);
 228
 229         if (del)
 230                 lfsck_layout_llst_put(llst);
 231 }
 232
 233 static inline struct lfsck_layout_slave_target *
 234 lfsck_layout_llst_find_and_del(struct lfsck_layout_slave_data *llsd,
 235                                __u32 index, bool unlink)
 236 {
 237         struct lfsck_layout_slave_target *llst;
 238
 239         spin_lock(&llsd->llsd_lock);
 240         list_for_each_entry(llst, &llsd->llsd_master_list, llst_list) {
 241                 if (llst->llst_index == index) {
 242                         if (unlink)
 243                                 list_del_init(&llst->llst_list);
 244                         else
 245                                 atomic_inc(&llst->llst_ref);
 246                         spin_unlock(&llsd->llsd_lock);
 247
 248                         return llst;
 249                 }
 250         }
 251         spin_unlock(&llsd->llsd_lock);
 252
 253         return NULL;
 254 }
 255
 256 static inline void lfsck_layout_object_put(const struct lu_env *env,
 257                                            struct lfsck_layout_object *llo)
 258 {
 259         if (atomic_dec_and_test(&llo->llo_ref)) {
 260                 lfsck_object_put(env, llo->llo_obj);
 261                 OBD_FREE_PTR(llo);
 262         }
 263 }
 264
 265 static struct lfsck_layout_req *
 266 lfsck_layout_req_init(struct lfsck_layout_object *parent,
 267                       struct dt_object *child, __u32 ost_idx, __u32 lov_idx)
 268 {
 269         struct lfsck_layout_req *llr;
 270
 271         OBD_ALLOC_PTR(llr);
 272         if (llr == NULL)
 273                 return ERR_PTR(-ENOMEM);
 274
 275         INIT_LIST_HEAD(&llr->llr_list);
 276         atomic_inc(&parent->llo_ref);
 277         llr->llr_parent = parent;
 278         llr->llr_child = child;
 279         llr->llr_ost_idx = ost_idx;
 280         llr->llr_lov_idx = lov_idx;
 281
 282         return llr;
 283 }
 284
 285 static inline void lfsck_layout_req_fini(const struct lu_env *env,
 286                                          struct lfsck_layout_req *llr)
 287 {
 288         lu_object_put(env, &llr->llr_child->do_lu);
 289         lfsck_layout_object_put(env, llr->llr_parent);
 290         OBD_FREE_PTR(llr);
 291 }
 292
 293 static inline bool lfsck_layout_req_empty(struct lfsck_layout_master_data *llmd)
 294 {
 295         bool empty = false;
 296
 297         spin_lock(&llmd->llmd_lock);
 298         if (list_empty(&llmd->llmd_req_list))
 299                 empty = true;
 300         spin_unlock(&llmd->llmd_lock);
 301
 302         return empty;
 303 }
 304
 305 static int lfsck_layout_get_lovea(const struct lu_env *env,
 306                                   struct dt_object *obj,
 307                                   struct lu_buf *buf, ssize_t *buflen)
 308 {
 309         int rc;
 310
 311 again:
 312         rc = dt_xattr_get(env, obj, buf, XATTR_NAME_LOV, BYPASS_CAPA);
 313         if (rc == -ERANGE) {
 314                 rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_LOV,
 315                                   BYPASS_CAPA);
 316                 if (rc <= 0)
 317                         return rc;
 318
 319                 lu_buf_realloc(buf, rc);
 320                 if (buflen != NULL)
 321                         *buflen = buf->lb_len;
 322
 323                 if (buf->lb_buf == NULL)
 324                         return -ENOMEM;
 325
 326                 goto again;
 327         }
 328
 329         if (rc == -ENODATA)
 330                 rc = 0;
 331
 332         if (rc <= 0)
 333                 return rc;
 334
 335         if (unlikely(buf->lb_buf == NULL)) {
 336                 lu_buf_alloc(buf, rc);
 337                 if (buflen != NULL)
 338                         *buflen = buf->lb_len;
 339
 340                 if (buf->lb_buf == NULL)
 341                         return -ENOMEM;
 342
 343                 goto again;
 344         }
 345
 346         return rc;
 347 }
 348
 349 static int lfsck_layout_verify_header(struct lov_mds_md_v1 *lmm)
 350 {
 351         __u32 magic;
 352         __u32 patten;
 353
 354         magic = le32_to_cpu(lmm->lmm_magic);
 355         /* If magic crashed, keep it there. Sometime later, during OST-object
 356          * orphan handling, if some OST-object(s) back-point to it, it can be
 357          * verified and repaired. */
 358         if (magic != LOV_MAGIC_V1 && magic != LOV_MAGIC_V3)
 359                 return -EINVAL;
 360
 361         patten = le32_to_cpu(lmm->lmm_pattern);
 362         /* XXX: currently, we only support LOV_PATTERN_RAID0. */
 363         if (patten != LOV_PATTERN_RAID0)
 364                 return -EOPNOTSUPP;
 365
 366         return 0;
 367 }
 368
 369 #define LFSCK_RBTREE_BITMAP_SIZE        PAGE_CACHE_SIZE
 370 #define LFSCK_RBTREE_BITMAP_WIDTH       (LFSCK_RBTREE_BITMAP_SIZE << 3)
 371 #define LFSCK_RBTREE_BITMAP_MASK        (LFSCK_RBTREE_BITMAP_SIZE - 1)
 372
 373 struct lfsck_rbtree_node {
 374         struct rb_node   lrn_node;
 375         __u64            lrn_seq;
 376         __u32            lrn_first_oid;
 377         atomic_t         lrn_known_count;
 378         atomic_t         lrn_accessed_count;
 379         void            *lrn_known_bitmap;
 380         void            *lrn_accessed_bitmap;
 381 };
 382
 383 static inline int lfsck_rbtree_cmp(struct lfsck_rbtree_node *lrn,
 384                                    __u64 seq, __u32 oid)
 385 {
 386         if (seq < lrn->lrn_seq)
 387                 return -1;
 388
 389         if (seq > lrn->lrn_seq)
 390                 return 1;
 391
 392         if (oid < lrn->lrn_first_oid)
 393                 return -1;
 394
 395         if (oid >= lrn->lrn_first_oid + LFSCK_RBTREE_BITMAP_WIDTH)
 396                 return 1;
 397
 398         return 0;
 399 }
 400
 401 /* The caller should hold llsd->llsd_rb_lock. */
 402 static struct lfsck_rbtree_node *
 403 lfsck_rbtree_search(struct lfsck_layout_slave_data *llsd,
 404                     const struct lu_fid *fid, bool *exact)
 405 {
 406         struct rb_node           *node  = llsd->llsd_rb_root.rb_node;
 407         struct rb_node           *prev  = NULL;
 408         struct lfsck_rbtree_node *lrn   = NULL;
 409         int                       rc    = 0;
 410
 411         if (exact != NULL)
 412                 *exact = true;
 413
 414         while (node != NULL) {
 415                 prev = node;
 416                 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
 417                 rc = lfsck_rbtree_cmp(lrn, fid_seq(fid), fid_oid(fid));
 418                 if (rc < 0)
 419                         node = node->rb_left;
 420                 else if (rc > 0)
 421                         node = node->rb_right;
 422                 else
 423                         return lrn;
 424         }
 425
 426         if (exact == NULL)
 427                 return NULL;
 428
 429         /* If there is no exactly matched one, then to the next valid one. */
 430         *exact = false;
 431
 432         /* The rbtree is empty. */
 433         if (rc == 0)
 434                 return NULL;
 435
 436         if (rc < 0)
 437                 return lrn;
 438
 439         node = rb_next(prev);
 440
 441         /* The end of the rbtree. */
 442         if (node == NULL)
 443                 return NULL;
 444
 445         lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
 446
 447         return lrn;
 448 }
 449
 450 static struct lfsck_rbtree_node *lfsck_rbtree_new(const struct lu_env *env,
 451                                                   const struct lu_fid *fid)
 452 {
 453         struct lfsck_rbtree_node *lrn;
 454
 455         OBD_ALLOC_PTR(lrn);
 456         if (lrn == NULL)
 457                 return ERR_PTR(-ENOMEM);
 458
 459         OBD_ALLOC(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
 460         if (lrn->lrn_known_bitmap == NULL) {
 461                 OBD_FREE_PTR(lrn);
 462
 463                 return ERR_PTR(-ENOMEM);
 464         }
 465
 466         OBD_ALLOC(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
 467         if (lrn->lrn_accessed_bitmap == NULL) {
 468                 OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
 469                 OBD_FREE_PTR(lrn);
 470
 471                 return ERR_PTR(-ENOMEM);
 472         }
 473
 474         rb_init_node(&lrn->lrn_node);
 475         lrn->lrn_seq = fid_seq(fid);
 476         lrn->lrn_first_oid = fid_oid(fid) & ~LFSCK_RBTREE_BITMAP_MASK;
 477         atomic_set(&lrn->lrn_known_count, 0);
 478         atomic_set(&lrn->lrn_accessed_count, 0);
 479
 480         return lrn;
 481 }
 482
 483 static void lfsck_rbtree_free(struct lfsck_rbtree_node *lrn)
 484 {
 485         OBD_FREE(lrn->lrn_accessed_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
 486         OBD_FREE(lrn->lrn_known_bitmap, LFSCK_RBTREE_BITMAP_SIZE);
 487         OBD_FREE_PTR(lrn);
 488 }
 489
 490 /* The caller should hold lock. */
 491 static struct lfsck_rbtree_node *
 492 lfsck_rbtree_insert(struct lfsck_layout_slave_data *llsd,
 493                     struct lfsck_rbtree_node *lrn)
 494 {
 495         struct rb_node           **pos    = &(llsd->llsd_rb_root.rb_node);
 496         struct rb_node            *parent = NULL;
 497         struct lfsck_rbtree_node  *tmp;
 498         int                        rc;
 499
 500         while (*pos) {
 501                 parent = *pos;
 502                 tmp = rb_entry(*pos, struct lfsck_rbtree_node, lrn_node);
 503                 rc = lfsck_rbtree_cmp(tmp, lrn->lrn_seq, lrn->lrn_first_oid);
 504                 if (rc < 0)
 505                         pos = &((*pos)->rb_left);
 506                 else if (rc > 0)
 507                         pos = &((*pos)->rb_right);
 508                 else
 509                         return tmp;
 510         }
 511
 512         rb_link_node(&lrn->lrn_node, parent, pos);
 513         rb_insert_color(&lrn->lrn_node, &llsd->llsd_rb_root);
 514
 515         return lrn;
 516 }
 517
 518 extern const struct dt_index_operations lfsck_orphan_index_ops;
 519
 520 static int lfsck_rbtree_setup(const struct lu_env *env,
 521                               struct lfsck_component *com)
 522 {
 523         struct lu_fid                   *fid    = &lfsck_env_info(env)->lti_fid;
 524         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 525         struct dt_device                *dev    = lfsck->li_bottom;
 526         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
 527         struct dt_object                *obj;
 528
 529         fid->f_seq = FID_SEQ_LAYOUT_RBTREE;
 530         fid->f_oid = lfsck_dev_idx(dev);
 531         fid->f_ver = 0;
 532         obj = dt_locate(env, dev, fid);
 533         if (IS_ERR(obj))
 534                 RETURN(PTR_ERR(obj));
 535
 536         /* Generate an in-RAM object to stand for the layout rbtree.
 537          * Scanning the layout rbtree will be via the iteration over
 538          * the object. In the future, the rbtree may be written onto
 539          * disk with the object.
 540          *
 541          * Mark the object to be as exist. */
 542         obj->do_lu.lo_header->loh_attr |= LOHA_EXISTS;
 543         obj->do_index_ops = &lfsck_orphan_index_ops;
 544         llsd->llsd_rb_obj = obj;
 545         llsd->llsd_rbtree_valid = 1;
 546         dev->dd_record_fid_accessed = 1;
 547
 548         return 0;
 549 }
 550
 551 static void lfsck_rbtree_cleanup(const struct lu_env *env,
 552                                  struct lfsck_component *com)
 553 {
 554         struct lfsck_instance           *lfsck = com->lc_lfsck;
 555         struct lfsck_layout_slave_data  *llsd  = com->lc_data;
 556         struct rb_node                  *node  = rb_first(&llsd->llsd_rb_root);
 557         struct rb_node                  *next;
 558         struct lfsck_rbtree_node        *lrn;
 559
 560         lfsck->li_bottom->dd_record_fid_accessed = 0;
 561         /* Invalid the rbtree, then no others will use it. */
 562         write_lock(&llsd->llsd_rb_lock);
 563         llsd->llsd_rbtree_valid = 0;
 564         write_unlock(&llsd->llsd_rb_lock);
 565
 566         while (node != NULL) {
 567                 next = rb_next(node);
 568                 lrn = rb_entry(node, struct lfsck_rbtree_node, lrn_node);
 569                 rb_erase(node, &llsd->llsd_rb_root);
 570                 lfsck_rbtree_free(lrn);
 571                 node = next;
 572         }
 573
 574         if (llsd->llsd_rb_obj != NULL) {
 575                 lu_object_put(env, &llsd->llsd_rb_obj->do_lu);
 576                 llsd->llsd_rb_obj = NULL;
 577         }
 578 }
 579
 580 static void lfsck_rbtree_update_bitmap(const struct lu_env *env,
 581                                        struct lfsck_component *com,
 582                                        const struct lu_fid *fid,
 583                                        bool accessed)
 584 {
 585         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
 586         struct lfsck_rbtree_node        *lrn;
 587         bool                             insert = false;
 588         int                              idx;
 589         int                              rc     = 0;
 590         ENTRY;
 591
 592         CDEBUG(D_LFSCK, "%s: update bitmap for "DFID"\n",
 593                lfsck_lfsck2name(com->lc_lfsck), PFID(fid));
 594
 595         if (unlikely(!fid_is_sane(fid) || fid_is_last_id(fid)))
 596                 RETURN_EXIT;
 597
 598         if (!fid_is_idif(fid) && !fid_is_norm(fid))
 599                 RETURN_EXIT;
 600
 601         read_lock(&llsd->llsd_rb_lock);
 602         if (!llsd->llsd_rbtree_valid)
 603                 GOTO(unlock, rc = 0);
 604
 605         lrn = lfsck_rbtree_search(llsd, fid, NULL);
 606         if (lrn == NULL) {
 607                 struct lfsck_rbtree_node *tmp;
 608
 609                 LASSERT(!insert);
 610
 611                 read_unlock(&llsd->llsd_rb_lock);
 612                 tmp = lfsck_rbtree_new(env, fid);
 613                 if (IS_ERR(tmp))
 614                         GOTO(out, rc = PTR_ERR(tmp));
 615
 616                 insert = true;
 617                 write_lock(&llsd->llsd_rb_lock);
 618                 if (!llsd->llsd_rbtree_valid) {
 619                         lfsck_rbtree_free(tmp);
 620                         GOTO(unlock, rc = 0);
 621                 }
 622
 623                 lrn = lfsck_rbtree_insert(llsd, tmp);
 624                 if (lrn != tmp)
 625                         lfsck_rbtree_free(tmp);
 626         }
 627
 628         idx = fid_oid(fid) & LFSCK_RBTREE_BITMAP_MASK;
 629         /* Any accessed object must be a known object. */
 630         if (!test_and_set_bit(idx, lrn->lrn_known_bitmap))
 631                 atomic_inc(&lrn->lrn_known_count);
 632         if (accessed && !test_and_set_bit(idx, lrn->lrn_accessed_bitmap))
 633                 atomic_inc(&lrn->lrn_accessed_count);
 634
 635         GOTO(unlock, rc = 0);
 636
 637 unlock:
 638         if (insert)
 639                 write_unlock(&llsd->llsd_rb_lock);
 640         else
 641                 read_unlock(&llsd->llsd_rb_lock);
 642 out:
 643         if (rc != 0 && accessed) {
 644                 struct lfsck_layout *lo = com->lc_file_ram;
 645
 646                 CERROR("%s: Fail to update object accessed bitmap, will cause "
 647                        "incorrect LFSCK OST-object handling, so disable it to "
 648                        "cancel orphan handling for related device. rc = %d.\n",
 649                        lfsck_lfsck2name(com->lc_lfsck), rc);
 650                 lo->ll_flags |= LF_INCOMPLETE;
 651                 lfsck_rbtree_cleanup(env, com);
 652         }
 653 }
 654
 655 static inline bool is_dummy_lov_ost_data(struct lov_ost_data_v1 *obj)
 656 {
 657         if (fid_is_zero(&obj->l_ost_oi.oi_fid) &&
 658             obj->l_ost_gen == 0 && obj->l_ost_idx == 0)
 659                 return true;
 660
 661         return false;
 662 }
 663
 664 static void lfsck_layout_le_to_cpu(struct lfsck_layout *des,
 665                                    const struct lfsck_layout *src)
 666 {
 667         int i;
 668
 669         des->ll_magic = le32_to_cpu(src->ll_magic);
 670         des->ll_status = le32_to_cpu(src->ll_status);
 671         des->ll_flags = le32_to_cpu(src->ll_flags);
 672         des->ll_success_count = le32_to_cpu(src->ll_success_count);
 673         des->ll_run_time_phase1 = le32_to_cpu(src->ll_run_time_phase1);
 674         des->ll_run_time_phase2 = le32_to_cpu(src->ll_run_time_phase2);
 675         des->ll_time_last_complete = le64_to_cpu(src->ll_time_last_complete);
 676         des->ll_time_latest_start = le64_to_cpu(src->ll_time_latest_start);
 677         des->ll_time_last_checkpoint =
 678                                 le64_to_cpu(src->ll_time_last_checkpoint);
 679         des->ll_pos_latest_start = le64_to_cpu(src->ll_pos_latest_start);
 680         des->ll_pos_last_checkpoint = le64_to_cpu(src->ll_pos_last_checkpoint);
 681         des->ll_pos_first_inconsistent =
 682                         le64_to_cpu(src->ll_pos_first_inconsistent);
 683         des->ll_objs_checked_phase1 = le64_to_cpu(src->ll_objs_checked_phase1);
 684         des->ll_objs_failed_phase1 = le64_to_cpu(src->ll_objs_failed_phase1);
 685         des->ll_objs_checked_phase2 = le64_to_cpu(src->ll_objs_checked_phase2);
 686         des->ll_objs_failed_phase2 = le64_to_cpu(src->ll_objs_failed_phase2);
 687         for (i = 0; i < LLIT_MAX; i++)
 688                 des->ll_objs_repaired[i] =
 689                                 le64_to_cpu(src->ll_objs_repaired[i]);
 690         des->ll_objs_skipped = le64_to_cpu(src->ll_objs_skipped);
 691 }
 692
 693 static void lfsck_layout_cpu_to_le(struct lfsck_layout *des,
 694                                    const struct lfsck_layout *src)
 695 {
 696         int i;
 697
 698         des->ll_magic = cpu_to_le32(src->ll_magic);
 699         des->ll_status = cpu_to_le32(src->ll_status);
 700         des->ll_flags = cpu_to_le32(src->ll_flags);
 701         des->ll_success_count = cpu_to_le32(src->ll_success_count);
 702         des->ll_run_time_phase1 = cpu_to_le32(src->ll_run_time_phase1);
 703         des->ll_run_time_phase2 = cpu_to_le32(src->ll_run_time_phase2);
 704         des->ll_time_last_complete = cpu_to_le64(src->ll_time_last_complete);
 705         des->ll_time_latest_start = cpu_to_le64(src->ll_time_latest_start);
 706         des->ll_time_last_checkpoint =
 707                                 cpu_to_le64(src->ll_time_last_checkpoint);
 708         des->ll_pos_latest_start = cpu_to_le64(src->ll_pos_latest_start);
 709         des->ll_pos_last_checkpoint = cpu_to_le64(src->ll_pos_last_checkpoint);
 710         des->ll_pos_first_inconsistent =
 711                         cpu_to_le64(src->ll_pos_first_inconsistent);
 712         des->ll_objs_checked_phase1 = cpu_to_le64(src->ll_objs_checked_phase1);
 713         des->ll_objs_failed_phase1 = cpu_to_le64(src->ll_objs_failed_phase1);
 714         des->ll_objs_checked_phase2 = cpu_to_le64(src->ll_objs_checked_phase2);
 715         des->ll_objs_failed_phase2 = cpu_to_le64(src->ll_objs_failed_phase2);
 716         for (i = 0; i < LLIT_MAX; i++)
 717                 des->ll_objs_repaired[i] =
 718                                 cpu_to_le64(src->ll_objs_repaired[i]);
 719         des->ll_objs_skipped = cpu_to_le64(src->ll_objs_skipped);
 720 }
 721
 722 /**
 723  * \retval +ve: the lfsck_layout is broken, the caller should reset it.
 724  * \retval 0: succeed.
 725  * \retval -ve: failed cases.
 726  */
 727 static int lfsck_layout_load(const struct lu_env *env,
 728                              struct lfsck_component *com)
 729 {
 730         struct lfsck_layout             *lo     = com->lc_file_ram;
 731         const struct dt_body_operations *dbo    = com->lc_obj->do_body_ops;
 732         ssize_t                          size   = com->lc_file_size;
 733         loff_t                           pos    = 0;
 734         int                              rc;
 735
 736         rc = dbo->dbo_read(env, com->lc_obj,
 737                            lfsck_buf_get(env, com->lc_file_disk, size), &pos,
 738                            BYPASS_CAPA);
 739         if (rc == 0) {
 740                 return -ENOENT;
 741         } else if (rc < 0) {
 742                 CWARN("%s: failed to load lfsck_layout: rc = %d\n",
 743                       lfsck_lfsck2name(com->lc_lfsck), rc);
 744                 return rc;
 745         } else if (rc != size) {
 746                 CWARN("%s: crashed lfsck_layout, to be reset: rc = %d\n",
 747                       lfsck_lfsck2name(com->lc_lfsck), rc);
 748                 return 1;
 749         }
 750
 751         lfsck_layout_le_to_cpu(lo, com->lc_file_disk);
 752         if (lo->ll_magic != LFSCK_LAYOUT_MAGIC) {
 753                 CWARN("%s: invalid lfsck_layout magic %#x != %#x, "
 754                       "to be reset\n", lfsck_lfsck2name(com->lc_lfsck),
 755                       lo->ll_magic, LFSCK_LAYOUT_MAGIC);
 756                 return 1;
 757         }
 758
 759         return 0;
 760 }
 761
 762 static int lfsck_layout_store(const struct lu_env *env,
 763                               struct lfsck_component *com)
 764 {
 765         struct dt_object         *obj           = com->lc_obj;
 766         struct lfsck_instance    *lfsck         = com->lc_lfsck;
 767         struct lfsck_layout      *lo            = com->lc_file_disk;
 768         struct thandle           *handle;
 769         ssize_t                   size          = com->lc_file_size;
 770         loff_t                    pos           = 0;
 771         int                       rc;
 772         ENTRY;
 773
 774         lfsck_layout_cpu_to_le(lo, com->lc_file_ram);
 775         handle = dt_trans_create(env, lfsck->li_bottom);
 776         if (IS_ERR(handle)) {
 777                 rc = PTR_ERR(handle);
 778                 CERROR("%s: fail to create trans for storing lfsck_layout: "
 779                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 780                 RETURN(rc);
 781         }
 782
 783         rc = dt_declare_record_write(env, obj, size, pos, handle);
 784         if (rc != 0) {
 785                 CERROR("%s: fail to declare trans for storing lfsck_layout(1): "
 786                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 787                 GOTO(out, rc);
 788         }
 789
 790         rc = dt_trans_start_local(env, lfsck->li_bottom, handle);
 791         if (rc != 0) {
 792                 CERROR("%s: fail to start trans for storing lfsck_layout: "
 793                        "rc = %d\n", lfsck_lfsck2name(lfsck), rc);
 794                 GOTO(out, rc);
 795         }
 796
 797         rc = dt_record_write(env, obj, lfsck_buf_get(env, lo, size), &pos,
 798                              handle);
 799         if (rc != 0)
 800                 CERROR("%s: fail to store lfsck_layout(1): size = %d, "
 801                        "rc = %d\n", lfsck_lfsck2name(lfsck), (int)size, rc);
 802
 803         GOTO(out, rc);
 804
 805 out:
 806         dt_trans_stop(env, lfsck->li_bottom, handle);
 807
 808         return rc;
 809 }
 810
 811 static int lfsck_layout_init(const struct lu_env *env,
 812                              struct lfsck_component *com)
 813 {
 814         struct lfsck_layout *lo = com->lc_file_ram;
 815         int rc;
 816
 817         memset(lo, 0, com->lc_file_size);
 818         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
 819         lo->ll_status = LS_INIT;
 820         down_write(&com->lc_sem);
 821         rc = lfsck_layout_store(env, com);
 822         up_write(&com->lc_sem);
 823
 824         return rc;
 825 }
 826
 827 static int fid_is_for_ostobj(const struct lu_env *env, struct dt_device *dt,
 828                              struct dt_object *obj, const struct lu_fid *fid)
 829 {
 830         struct seq_server_site  *ss     = lu_site2seq(dt->dd_lu_dev.ld_site);
 831         struct lu_seq_range      range  = { 0 };
 832         struct lustre_mdt_attrs *lma;
 833         int                      rc;
 834
 835         fld_range_set_any(&range);
 836         rc = fld_server_lookup(env, ss->ss_server_fld, fid_seq(fid), &range);
 837         if (rc == 0) {
 838                 if (fld_range_is_ost(&range))
 839                         return 1;
 840
 841                 return 0;
 842         }
 843
 844         lma = &lfsck_env_info(env)->lti_lma;
 845         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, lma, sizeof(*lma)),
 846                           XATTR_NAME_LMA, BYPASS_CAPA);
 847         if (rc == sizeof(*lma)) {
 848                 lustre_lma_swab(lma);
 849
 850                 return lma->lma_compat & LMAC_FID_ON_OST ? 1 : 0;
 851         }
 852
 853         rc = dt_xattr_get(env, obj, &LU_BUF_NULL, XATTR_NAME_FID, BYPASS_CAPA);
 854
 855         return rc > 0;
 856 }
 857
 858 static struct lfsck_layout_seq *
 859 lfsck_layout_seq_lookup(struct lfsck_layout_slave_data *llsd, __u64 seq)
 860 {
 861         struct lfsck_layout_seq *lls;
 862
 863         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
 864                 if (lls->lls_seq == seq)
 865                         return lls;
 866
 867                 if (lls->lls_seq > seq)
 868                         return NULL;
 869         }
 870
 871         return NULL;
 872 }
 873
 874 static void
 875 lfsck_layout_seq_insert(struct lfsck_layout_slave_data *llsd,
 876                         struct lfsck_layout_seq *lls)
 877 {
 878         struct lfsck_layout_seq *tmp;
 879         struct list_head        *pos = &llsd->llsd_seq_list;
 880
 881         list_for_each_entry(tmp, &llsd->llsd_seq_list, lls_list) {
 882                 if (lls->lls_seq < tmp->lls_seq) {
 883                         pos = &tmp->lls_list;
 884                         break;
 885                 }
 886         }
 887         list_add_tail(&lls->lls_list, pos);
 888 }
 889
 890 static int
 891 lfsck_layout_lastid_create(const struct lu_env *env,
 892                            struct lfsck_instance *lfsck,
 893                            struct dt_object *obj)
 894 {
 895         struct lfsck_thread_info *info   = lfsck_env_info(env);
 896         struct lu_attr           *la     = &info->lti_la;
 897         struct dt_object_format  *dof    = &info->lti_dof;
 898         struct lfsck_bookmark    *bk     = &lfsck->li_bookmark_ram;
 899         struct dt_device         *dt     = lfsck->li_bottom;
 900         struct thandle           *th;
 901         __u64                     lastid = 0;
 902         loff_t                    pos    = 0;
 903         int                       rc;
 904         ENTRY;
 905
 906         CDEBUG(D_LFSCK, "To create LAST_ID for <seq> "LPX64"\n",
 907                fid_seq(lfsck_dto2fid(obj)));
 908
 909         if (bk->lb_param & LPF_DRYRUN)
 910                 return 0;
 911
 912         memset(la, 0, sizeof(*la));
 913         la->la_mode = S_IFREG |  S_IRUGO | S_IWUSR;
 914         la->la_valid = LA_MODE | LA_UID | LA_GID;
 915         dof->dof_type = dt_mode_to_dft(S_IFREG);
 916
 917         th = dt_trans_create(env, dt);
 918         if (IS_ERR(th))
 919                 RETURN(rc = PTR_ERR(th));
 920
 921         rc = dt_declare_create(env, obj, la, NULL, dof, th);
 922         if (rc != 0)
 923                 GOTO(stop, rc);
 924
 925         rc = dt_declare_record_write(env, obj, sizeof(lastid), pos, th);
 926         if (rc != 0)
 927                 GOTO(stop, rc);
 928
 929         rc = dt_trans_start_local(env, dt, th);
 930         if (rc != 0)
 931                 GOTO(stop, rc);
 932
 933         dt_write_lock(env, obj, 0);
 934         if (likely(!dt_object_exists(obj))) {
 935                 rc = dt_create(env, obj, la, NULL, dof, th);
 936                 if (rc == 0)
 937                         rc = dt_record_write(env, obj,
 938                                 lfsck_buf_get(env, &lastid, sizeof(lastid)),
 939                                 &pos, th);
 940         }
 941         dt_write_unlock(env, obj);
 942
 943         GOTO(stop, rc);
 944
 945 stop:
 946         dt_trans_stop(env, dt, th);
 947
 948         return rc;
 949 }
 950
 951 static int
 952 lfsck_layout_lastid_reload(const struct lu_env *env,
 953                            struct lfsck_component *com,
 954                            struct lfsck_layout_seq *lls)
 955 {
 956         __u64   lastid;
 957         loff_t  pos     = 0;
 958         int     rc;
 959
 960         dt_read_lock(env, lls->lls_lastid_obj, 0);
 961         rc = dt_record_read(env, lls->lls_lastid_obj,
 962                             lfsck_buf_get(env, &lastid, sizeof(lastid)), &pos);
 963         dt_read_unlock(env, lls->lls_lastid_obj);
 964         if (unlikely(rc != 0))
 965                 return rc;
 966
 967         lastid = le64_to_cpu(lastid);
 968         if (lastid < lls->lls_lastid_known) {
 969                 struct lfsck_instance   *lfsck  = com->lc_lfsck;
 970                 struct lfsck_layout     *lo     = com->lc_file_ram;
 971
 972                 lls->lls_lastid = lls->lls_lastid_known;
 973                 lls->lls_dirty = 1;
 974                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
 975                         LASSERT(lfsck->li_out_notify != NULL);
 976
 977                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
 978                                              LE_LASTID_REBUILDING);
 979                         lo->ll_flags |= LF_CRASHED_LASTID;
 980                 }
 981         } else if (lastid >= lls->lls_lastid) {
 982                 lls->lls_lastid = lastid;
 983                 lls->lls_dirty = 0;
 984         }
 985
 986         return 0;
 987 }
 988
 989 static int
 990 lfsck_layout_lastid_store(const struct lu_env *env,
 991                           struct lfsck_component *com)
 992 {
 993         struct lfsck_instance           *lfsck  = com->lc_lfsck;
 994         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
 995         struct dt_device                *dt     = lfsck->li_bottom;
 996         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
 997         struct lfsck_layout_seq         *lls;
 998         struct thandle                  *th;
 999         __u64                            lastid;
1000         int                              rc     = 0;
1001         int                              rc1    = 0;
1002
1003         list_for_each_entry(lls, &llsd->llsd_seq_list, lls_list) {
1004                 loff_t pos = 0;
1005
1006                 /* XXX: Add the code back if we really found related
1007                  *      inconsistent cases in the future. */
1008 #if 0
1009                 if (!lls->lls_dirty) {
1010                         /* In OFD, before the pre-creation, the LAST_ID
1011                          * file will be updated firstly, which may hide
1012                          * some potential crashed cases. For example:
1013                          *
1014                          * The old obj1's ID is higher than old LAST_ID
1015                          * but lower than the new LAST_ID, but the LFSCK
1016                          * have not touch the obj1 until the OFD updated
1017                          * the LAST_ID. So the LFSCK does not regard it
1018                          * as crashed case. But when OFD does not create
1019                          * successfully, it will set the LAST_ID as the
1020                          * real created objects' ID, then LFSCK needs to
1021                          * found related inconsistency. */
1022                         rc = lfsck_layout_lastid_reload(env, com, lls);
1023                         if (likely(!lls->lls_dirty))
1024                                 continue;
1025                 }
1026 #endif
1027
1028                 CDEBUG(D_LFSCK, "To sync the LAST_ID for <seq> "LPX64
1029                        " as <oid> "LPU64"\n", lls->lls_seq, lls->lls_lastid);
1030
1031                 if (bk->lb_param & LPF_DRYRUN) {
1032                         lls->lls_dirty = 0;
1033                         continue;
1034                 }
1035
1036                 th = dt_trans_create(env, dt);
1037                 if (IS_ERR(th)) {
1038                         rc1 = PTR_ERR(th);
1039                         CERROR("%s: (1) failed to store "LPX64": rc = %d\n",
1040                                lfsck_lfsck2name(com->lc_lfsck),
1041                                lls->lls_seq, rc1);
1042                         continue;
1043                 }
1044
1045                 rc = dt_declare_record_write(env, lls->lls_lastid_obj,
1046                                              sizeof(lastid), pos, th);
1047                 if (rc != 0)
1048                         goto stop;
1049
1050                 rc = dt_trans_start_local(env, dt, th);
1051                 if (rc != 0)
1052                         goto stop;
1053
1054                 lastid = cpu_to_le64(lls->lls_lastid);
1055                 dt_write_lock(env, lls->lls_lastid_obj, 0);
1056                 rc = dt_record_write(env, lls->lls_lastid_obj,
1057                                      lfsck_buf_get(env, &lastid,
1058                                      sizeof(lastid)), &pos, th);
1059                 dt_write_unlock(env, lls->lls_lastid_obj);
1060                 if (rc == 0)
1061                         lls->lls_dirty = 0;
1062
1063 stop:
1064                 dt_trans_stop(env, dt, th);
1065                 if (rc != 0) {
1066                         rc1 = rc;
1067                         CERROR("%s: (2) failed to store "LPX64": rc = %d\n",
1068                                lfsck_lfsck2name(com->lc_lfsck),
1069                                lls->lls_seq, rc1);
1070                 }
1071         }
1072
1073         return rc1;
1074 }
1075
1076 static int
1077 lfsck_layout_lastid_load(const struct lu_env *env,
1078                          struct lfsck_component *com,
1079                          struct lfsck_layout_seq *lls)
1080 {
1081         struct lfsck_instance   *lfsck  = com->lc_lfsck;
1082         struct lfsck_layout     *lo     = com->lc_file_ram;
1083         struct lu_fid           *fid    = &lfsck_env_info(env)->lti_fid;
1084         struct dt_object        *obj;
1085         loff_t                   pos    = 0;
1086         int                      rc;
1087         ENTRY;
1088
1089         lu_last_id_fid(fid, lls->lls_seq, lfsck_dev_idx(lfsck->li_bottom));
1090         obj = dt_locate(env, lfsck->li_bottom, fid);
1091         if (IS_ERR(obj))
1092                 RETURN(PTR_ERR(obj));
1093
1094         /* LAST_ID crashed, to be rebuilt */
1095         if (!dt_object_exists(obj)) {
1096                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
1097                         LASSERT(lfsck->li_out_notify != NULL);
1098
1099                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1100                                              LE_LASTID_REBUILDING);
1101                         lo->ll_flags |= LF_CRASHED_LASTID;
1102
1103                         if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY4) &&
1104                             cfs_fail_val > 0) {
1105                                 struct l_wait_info lwi = LWI_TIMEOUT(
1106                                                 cfs_time_seconds(cfs_fail_val),
1107                                                 NULL, NULL);
1108
1109                                 up_write(&com->lc_sem);
1110                                 l_wait_event(lfsck->li_thread.t_ctl_waitq,
1111                                              !thread_is_running(&lfsck->li_thread),
1112                                              &lwi);
1113                                 down_write(&com->lc_sem);
1114                         }
1115                 }
1116
1117                 rc = lfsck_layout_lastid_create(env, lfsck, obj);
1118         } else {
1119                 dt_read_lock(env, obj, 0);
1120                 rc = dt_read(env, obj,
1121                         lfsck_buf_get(env, &lls->lls_lastid, sizeof(__u64)),
1122                         &pos);
1123                 dt_read_unlock(env, obj);
1124                 if (rc != 0 && rc != sizeof(__u64))
1125                         GOTO(out, rc = (rc > 0 ? -EFAULT : rc));
1126
1127                 if (rc == 0 && !(lo->ll_flags & LF_CRASHED_LASTID)) {
1128                         LASSERT(lfsck->li_out_notify != NULL);
1129
1130                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
1131                                              LE_LASTID_REBUILDING);
1132                         lo->ll_flags |= LF_CRASHED_LASTID;
1133                 }
1134
1135                 lls->lls_lastid = le64_to_cpu(lls->lls_lastid);
1136                 rc = 0;
1137         }
1138
1139         GOTO(out, rc);
1140
1141 out:
1142         if (rc != 0)
1143                 lfsck_object_put(env, obj);
1144         else
1145                 lls->lls_lastid_obj = obj;
1146
1147         return rc;
1148 }
1149
1150 static int lfsck_layout_master_async_interpret(const struct lu_env *env,
1151                                                struct ptlrpc_request *req,
1152                                                void *args, int rc)
1153 {
1154         struct lfsck_async_interpret_args *laia = args;
1155         struct lfsck_component            *com  = laia->laia_com;
1156         struct lfsck_layout_master_data   *llmd = com->lc_data;
1157         struct lfsck_tgt_descs            *ltds = laia->laia_ltds;
1158         struct lfsck_tgt_desc             *ltd  = laia->laia_ltd;
1159         struct lfsck_request              *lr   = laia->laia_lr;
1160
1161         switch (lr->lr_event) {
1162         case LE_START:
1163                 if (rc != 0) {
1164                         struct lfsck_layout *lo = com->lc_file_ram;
1165
1166                         CERROR("%s: fail to notify %s %x for layout start: "
1167                                "rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
1168                                (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
1169                                ltd->ltd_index, rc);
1170                         lo->ll_flags |= LF_INCOMPLETE;
1171                         break;
1172                 }
1173
1174                 spin_lock(&ltds->ltd_lock);
1175                 if (ltd->ltd_dead || ltd->ltd_layout_done) {
1176                         spin_unlock(&ltds->ltd_lock);
1177                         break;
1178                 }
1179
1180                 if (lr->lr_flags & LEF_TO_OST) {
1181                         if (list_empty(&ltd->ltd_layout_list))
1182                                 list_add_tail(&ltd->ltd_layout_list,
1183                                               &llmd->llmd_ost_list);
1184                         if (list_empty(&ltd->ltd_layout_phase_list))
1185                                 list_add_tail(&ltd->ltd_layout_phase_list,
1186                                               &llmd->llmd_ost_phase1_list);
1187                 } else {
1188                         if (list_empty(&ltd->ltd_layout_list))
1189                                 list_add_tail(&ltd->ltd_layout_list,
1190                                               &llmd->llmd_mdt_list);
1191                         if (list_empty(&ltd->ltd_layout_phase_list))
1192                                 list_add_tail(&ltd->ltd_layout_phase_list,
1193                                               &llmd->llmd_mdt_phase1_list);
1194                 }
1195                 spin_unlock(&ltds->ltd_lock);
1196                 break;
1197         case LE_STOP:
1198         case LE_PHASE1_DONE:
1199         case LE_PHASE2_DONE:
1200         case LE_PEER_EXIT:
1201                 if (rc != 0 && rc != -EALREADY)
1202                         CWARN("%s: fail to notify %s %x for layout: "
1203                               "event = %d, rc = %d\n",
1204                               lfsck_lfsck2name(com->lc_lfsck),
1205                               (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
1206                               ltd->ltd_index, lr->lr_event, rc);
1207                 break;
1208         case LE_QUERY: {
1209                 struct lfsck_reply *reply;
1210
1211                 if (rc != 0) {
1212                         spin_lock(&ltds->ltd_lock);
1213                         list_del_init(&ltd->ltd_layout_phase_list);
1214                         list_del_init(&ltd->ltd_layout_list);
1215                         spin_unlock(&ltds->ltd_lock);
1216                         break;
1217                 }
1218
1219                 reply = req_capsule_server_get(&req->rq_pill,
1220                                                &RMF_LFSCK_REPLY);
1221                 if (reply == NULL) {
1222                         rc = -EPROTO;
1223                         CERROR("%s: invalid return value: rc = %d\n",
1224                                lfsck_lfsck2name(com->lc_lfsck), rc);
1225                         spin_lock(&ltds->ltd_lock);
1226                         list_del_init(&ltd->ltd_layout_phase_list);
1227                         list_del_init(&ltd->ltd_layout_list);
1228                         spin_unlock(&ltds->ltd_lock);
1229                         break;
1230                 }
1231
1232                 switch (reply->lr_status) {
1233                 case LS_SCANNING_PHASE1:
1234                         break;
1235                 case LS_SCANNING_PHASE2:
1236                         spin_lock(&ltds->ltd_lock);
1237                         list_del_init(&ltd->ltd_layout_phase_list);
1238                         if (ltd->ltd_dead || ltd->ltd_layout_done) {
1239                                 spin_unlock(&ltds->ltd_lock);
1240                                 break;
1241                         }
1242
1243                         if (lr->lr_flags & LEF_TO_OST)
1244                                 list_add_tail(&ltd->ltd_layout_phase_list,
1245                                               &llmd->llmd_ost_phase2_list);
1246                         else
1247                                 list_add_tail(&ltd->ltd_layout_phase_list,
1248                                               &llmd->llmd_mdt_phase2_list);
1249                         spin_unlock(&ltds->ltd_lock);
1250                         break;
1251                 default:
1252                         spin_lock(&ltds->ltd_lock);
1253                         list_del_init(&ltd->ltd_layout_phase_list);
1254                         list_del_init(&ltd->ltd_layout_list);
1255                         spin_unlock(&ltds->ltd_lock);
1256                         break;
1257                 }
1258                 break;
1259         }
1260         default:
1261                 CERROR("%s: unexpected event: rc = %d\n",
1262                        lfsck_lfsck2name(com->lc_lfsck), lr->lr_event);
1263                 break;
1264         }
1265
1266         if (!laia->laia_shared) {
1267                 lfsck_tgt_put(ltd);
1268                 lfsck_component_put(env, com);
1269         }
1270
1271         return 0;
1272 }
1273
1274 static int lfsck_layout_master_query_others(const struct lu_env *env,
1275                                             struct lfsck_component *com)
1276 {
1277         struct lfsck_thread_info          *info  = lfsck_env_info(env);
1278         struct lfsck_request              *lr    = &info->lti_lr;
1279         struct lfsck_async_interpret_args *laia  = &info->lti_laia;
1280         struct lfsck_instance             *lfsck = com->lc_lfsck;
1281         struct lfsck_layout_master_data   *llmd  = com->lc_data;
1282         struct ptlrpc_request_set         *set;
1283         struct lfsck_tgt_descs            *ltds;
1284         struct lfsck_tgt_desc             *ltd;
1285         struct list_head                  *head;
1286         int                                rc    = 0;
1287         int                                rc1   = 0;
1288         ENTRY;
1289
1290         set = ptlrpc_prep_set();
1291         if (set == NULL)
1292                 RETURN(-ENOMEM);
1293
1294         llmd->llmd_touch_gen++;
1295         memset(lr, 0, sizeof(*lr));
1296         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
1297         lr->lr_event = LE_QUERY;
1298         lr->lr_active = LT_LAYOUT;
1299         laia->laia_com = com;
1300         laia->laia_lr = lr;
1301         laia->laia_shared = 0;
1302
1303         if (!list_empty(&llmd->llmd_mdt_phase1_list)) {
1304                 ltds = &lfsck->li_mdt_descs;
1305                 lr->lr_flags = 0;
1306                 head = &llmd->llmd_mdt_phase1_list;
1307         } else {
1308
1309 again:
1310                 ltds = &lfsck->li_ost_descs;
1311                 lr->lr_flags = LEF_TO_OST;
1312                 head = &llmd->llmd_ost_phase1_list;
1313         }
1314
1315         laia->laia_ltds = ltds;
1316         spin_lock(&ltds->ltd_lock);
1317         while (!list_empty(head)) {
1318                 ltd = list_entry(head->next,
1319                                  struct lfsck_tgt_desc,
1320                                  ltd_layout_phase_list);
1321                 if (ltd->ltd_layout_gen == llmd->llmd_touch_gen)
1322                         break;
1323
1324                 ltd->ltd_layout_gen = llmd->llmd_touch_gen;
1325                 list_del(&ltd->ltd_layout_phase_list);
1326                 list_add_tail(&ltd->ltd_layout_phase_list, head);
1327                 atomic_inc(&ltd->ltd_ref);
1328                 laia->laia_ltd = ltd;
1329                 spin_unlock(&ltds->ltd_lock);
1330                 rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
1331                                          lfsck_layout_master_async_interpret,
1332                                          laia, LFSCK_QUERY);
1333                 if (rc != 0) {
1334                         CERROR("%s: fail to query %s %x for layout: rc = %d\n",
1335                                lfsck_lfsck2name(lfsck),
1336                                (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
1337                                ltd->ltd_index, rc);
1338                         lfsck_tgt_put(ltd);
1339                         rc1 = rc;
1340                 }
1341                 spin_lock(&ltds->ltd_lock);
1342         }
1343         spin_unlock(&ltds->ltd_lock);
1344
1345         rc = ptlrpc_set_wait(set);
1346         if (rc < 0) {
1347                 ptlrpc_set_destroy(set);
1348                 RETURN(rc);
1349         }
1350
1351         if (!(lr->lr_flags & LEF_TO_OST) &&
1352             list_empty(&llmd->llmd_mdt_phase1_list))
1353                 goto again;
1354
1355         ptlrpc_set_destroy(set);
1356
1357         RETURN(rc1 != 0 ? rc1 : rc);
1358 }
1359
1360 static inline bool
1361 lfsck_layout_master_to_orphan(struct lfsck_layout_master_data *llmd)
1362 {
1363         return list_empty(&llmd->llmd_mdt_phase1_list) &&
1364                (!list_empty(&llmd->llmd_ost_phase2_list) ||
1365                 list_empty(&llmd->llmd_ost_phase1_list));
1366 }
1367
1368 static int lfsck_layout_master_notify_others(const struct lu_env *env,
1369                                              struct lfsck_component *com,
1370                                              struct lfsck_request *lr)
1371 {
1372         struct lfsck_thread_info          *info  = lfsck_env_info(env);
1373         struct lfsck_async_interpret_args *laia  = &info->lti_laia;
1374         struct lfsck_instance             *lfsck = com->lc_lfsck;
1375         struct lfsck_layout_master_data   *llmd  = com->lc_data;
1376         struct lfsck_layout               *lo    = com->lc_file_ram;
1377         struct lfsck_bookmark             *bk    = &lfsck->li_bookmark_ram;
1378         struct ptlrpc_request_set         *set;
1379         struct lfsck_tgt_descs            *ltds;
1380         struct lfsck_tgt_desc             *ltd;
1381         struct lfsck_tgt_desc             *next;
1382         struct list_head                  *head;
1383         __u32                              idx;
1384         int                                rc    = 0;
1385         ENTRY;
1386
1387         set = ptlrpc_prep_set();
1388         if (set == NULL)
1389                 RETURN(-ENOMEM);
1390
1391         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
1392         lr->lr_active = LT_LAYOUT;
1393         laia->laia_com = com;
1394         laia->laia_lr = lr;
1395         laia->laia_shared = 0;
1396         switch (lr->lr_event) {
1397         case LE_START:
1398                 /* Notify OSTs firstly, then handle other MDTs if needed. */
1399                 ltds = &lfsck->li_ost_descs;
1400                 laia->laia_ltds = ltds;
1401                 down_read(&ltds->ltd_rw_sem);
1402                 cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
1403                         ltd = lfsck_tgt_get(ltds, idx);
1404                         LASSERT(ltd != NULL);
1405
1406                         laia->laia_ltd = ltd;
1407                         ltd->ltd_layout_done = 0;
1408                         rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
1409                                         lfsck_layout_master_async_interpret,
1410                                         laia, LFSCK_NOTIFY);
1411                         if (rc != 0) {
1412                                 CERROR("%s: fail to notify %s %x for layout "
1413                                        "start: rc = %d\n",
1414                                        lfsck_lfsck2name(lfsck),
1415                                        (lr->lr_flags & LEF_TO_OST) ? "OST" :
1416                                        "MDT", idx, rc);
1417                                 lfsck_tgt_put(ltd);
1418                                 lo->ll_flags |= LF_INCOMPLETE;
1419                         }
1420                 }
1421                 up_read(&ltds->ltd_rw_sem);
1422
1423                 /* Sync up */
1424                 rc = ptlrpc_set_wait(set);
1425                 if (rc < 0) {
1426                         ptlrpc_set_destroy(set);
1427                         RETURN(rc);
1428                 }
1429
1430                 if (!(bk->lb_param & LPF_ALL_TGT))
1431                         break;
1432
1433                 /* link other MDT targets locallly. */
1434                 spin_lock(&ltds->ltd_lock);
1435                 cfs_foreach_bit(ltds->ltd_tgts_bitmap, idx) {
1436                         ltd = LTD_TGT(ltds, idx);
1437                         LASSERT(ltd != NULL);
1438
1439                         if (!list_empty(&ltd->ltd_layout_list))
1440                                 continue;
1441
1442                         list_add_tail(&ltd->ltd_layout_list,
1443                                       &llmd->llmd_mdt_list);
1444                         list_add_tail(&ltd->ltd_layout_phase_list,
1445                                       &llmd->llmd_mdt_phase1_list);
1446                 }
1447                 spin_unlock(&ltds->ltd_lock);
1448                 break;
1449         case LE_STOP:
1450         case LE_PHASE2_DONE:
1451         case LE_PEER_EXIT: {
1452                 /* Handle other MDTs firstly if needed, then notify the OSTs. */
1453                 if (bk->lb_param & LPF_ALL_TGT) {
1454                         head = &llmd->llmd_mdt_list;
1455                         ltds = &lfsck->li_mdt_descs;
1456                         if (lr->lr_event == LE_STOP) {
1457                                 /* unlink other MDT targets locallly. */
1458                                 spin_lock(&ltds->ltd_lock);
1459                                 list_for_each_entry_safe(ltd, next, head,
1460                                                          ltd_layout_list) {
1461                                         list_del_init(&ltd->ltd_layout_phase_list);
1462                                         list_del_init(&ltd->ltd_layout_list);
1463                                 }
1464                                 spin_unlock(&ltds->ltd_lock);
1465
1466                                 lr->lr_flags |= LEF_TO_OST;
1467                                 head = &llmd->llmd_ost_list;
1468                                 ltds = &lfsck->li_ost_descs;
1469                         } else {
1470                                 lr->lr_flags &= ~LEF_TO_OST;
1471                         }
1472                 } else {
1473                         lr->lr_flags |= LEF_TO_OST;
1474                         head = &llmd->llmd_ost_list;
1475                         ltds = &lfsck->li_ost_descs;
1476                 }
1477
1478 again:
1479                 laia->laia_ltds = ltds;
1480                 spin_lock(&ltds->ltd_lock);
1481                 while (!list_empty(head)) {
1482                         ltd = list_entry(head->next, struct lfsck_tgt_desc,
1483                                          ltd_layout_list);
1484                         if (!list_empty(&ltd->ltd_layout_phase_list))
1485                                 list_del_init(&ltd->ltd_layout_phase_list);
1486                         list_del_init(&ltd->ltd_layout_list);
1487                         atomic_inc(&ltd->ltd_ref);
1488                         laia->laia_ltd = ltd;
1489                         spin_unlock(&ltds->ltd_lock);
1490                         rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
1491                                         lfsck_layout_master_async_interpret,
1492                                         laia, LFSCK_NOTIFY);
1493                         if (rc != 0) {
1494                                 CERROR("%s: fail to notify %s %x for layout "
1495                                        "stop/phase2: rc = %d\n",
1496                                        lfsck_lfsck2name(lfsck),
1497                                        (lr->lr_flags & LEF_TO_OST) ? "OST" :
1498                                        "MDT", ltd->ltd_index, rc);
1499                                 lfsck_tgt_put(ltd);
1500                         }
1501                         spin_lock(&ltds->ltd_lock);
1502                 }
1503                 spin_unlock(&ltds->ltd_lock);
1504
1505                 rc = ptlrpc_set_wait(set);
1506                 if (rc < 0) {
1507                         ptlrpc_set_destroy(set);
1508                         RETURN(rc);
1509                 }
1510
1511                 if (!(lr->lr_flags & LEF_TO_OST)) {
1512                         lr->lr_flags |= LEF_TO_OST;
1513                         head = &llmd->llmd_ost_list;
1514                         ltds = &lfsck->li_ost_descs;
1515                         goto again;
1516                 }
1517                 break;
1518         }
1519         case LE_PHASE1_DONE:
1520                 llmd->llmd_touch_gen++;
1521                 ltds = &lfsck->li_mdt_descs;
1522                 laia->laia_ltds = ltds;
1523                 spin_lock(&ltds->ltd_lock);
1524                 while (!list_empty(&llmd->llmd_mdt_phase1_list)) {
1525                         ltd = list_entry(llmd->llmd_mdt_phase1_list.next,
1526                                          struct lfsck_tgt_desc,
1527                                          ltd_layout_phase_list);
1528                         if (ltd->ltd_layout_gen == llmd->llmd_touch_gen)
1529                                 break;
1530
1531                         ltd->ltd_layout_gen = llmd->llmd_touch_gen;
1532                         list_del_init(&ltd->ltd_layout_phase_list);
1533                         list_add_tail(&ltd->ltd_layout_phase_list,
1534                                       &llmd->llmd_mdt_phase1_list);
1535                         atomic_inc(&ltd->ltd_ref);
1536                         laia->laia_ltd = ltd;
1537                         spin_unlock(&ltds->ltd_lock);
1538                         rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
1539                                         lfsck_layout_master_async_interpret,
1540                                         laia, LFSCK_NOTIFY);
1541                         if (rc != 0) {
1542                                 CERROR("%s: fail to notify MDT %x for layout "
1543                                        "phase1 done: rc = %d\n",
1544                                        lfsck_lfsck2name(lfsck),
1545                                        ltd->ltd_index, rc);
1546                                 lfsck_tgt_put(ltd);
1547                         }
1548                         spin_lock(&ltds->ltd_lock);
1549                 }
1550                 spin_unlock(&ltds->ltd_lock);
1551                 break;
1552         default:
1553                 CERROR("%s: unexpected LFSCK event: rc = %d\n",
1554                        lfsck_lfsck2name(lfsck), lr->lr_event);
1555                 rc = -EINVAL;
1556                 break;
1557         }
1558
1559         rc = ptlrpc_set_wait(set);
1560         ptlrpc_set_destroy(set);
1561
1562         RETURN(rc);
1563 }
1564
1565 static int lfsck_layout_double_scan_result(const struct lu_env *env,
1566                                            struct lfsck_component *com,
1567                                            int rc)
1568 {
1569         struct lfsck_instance   *lfsck = com->lc_lfsck;
1570         struct lfsck_layout     *lo    = com->lc_file_ram;
1571         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
1572
1573         down_write(&com->lc_sem);
1574
1575         lo->ll_run_time_phase2 += cfs_duration_sec(cfs_time_current() +
1576                                 HALF_SEC - lfsck->li_time_last_checkpoint);
1577         lo->ll_time_last_checkpoint = cfs_time_current_sec();
1578         lo->ll_objs_checked_phase2 += com->lc_new_checked;
1579
1580         if (rc > 0) {
1581                 com->lc_journal = 0;
1582                 if (lo->ll_flags & LF_INCOMPLETE)
1583                         lo->ll_status = LS_PARTIAL;
1584                 else
1585                         lo->ll_status = LS_COMPLETED;
1586                 if (!(bk->lb_param & LPF_DRYRUN))
1587                         lo->ll_flags &= ~(LF_SCANNED_ONCE | LF_INCONSISTENT);
1588                 lo->ll_time_last_complete = lo->ll_time_last_checkpoint;
1589                 lo->ll_success_count++;
1590         } else if (rc == 0) {
1591                 lo->ll_status = lfsck->li_status;
1592                 if (lo->ll_status == 0)
1593                         lo->ll_status = LS_STOPPED;
1594         } else {
1595                 lo->ll_status = LS_FAILED;
1596         }
1597
1598         if (lo->ll_status != LS_PAUSED) {
1599                 spin_lock(&lfsck->li_lock);
1600                 list_del_init(&com->lc_link);
1601                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
1602                 spin_unlock(&lfsck->li_lock);
1603         }
1604
1605         rc = lfsck_layout_store(env, com);
1606
1607         up_write(&com->lc_sem);
1608
1609         return rc;
1610 }
1611
1612 static int lfsck_layout_lock(const struct lu_env *env,
1613                              struct lfsck_component *com,
1614                              struct dt_object *obj,
1615                              struct lustre_handle *lh, __u64 bits)
1616 {
1617         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1618         ldlm_policy_data_t              *policy = &info->lti_policy;
1619         struct ldlm_res_id              *resid  = &info->lti_resid;
1620         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1621         __u64                            flags  = LDLM_FL_ATOMIC_CB;
1622         int                              rc;
1623
1624         LASSERT(lfsck->li_namespace != NULL);
1625
1626         memset(policy, 0, sizeof(*policy));
1627         policy->l_inodebits.bits = bits;
1628         fid_build_reg_res_name(lfsck_dto2fid(obj), resid);
1629         rc = ldlm_cli_enqueue_local(lfsck->li_namespace, resid, LDLM_IBITS,
1630                                     policy, LCK_EX, &flags, ldlm_blocking_ast,
1631                                     ldlm_completion_ast, NULL, NULL, 0,
1632                                     LVB_T_NONE, NULL, lh);
1633         if (rc == ELDLM_OK) {
1634                 rc = 0;
1635         } else {
1636                 memset(lh, 0, sizeof(*lh));
1637                 rc = -EIO;
1638         }
1639
1640         return rc;
1641 }
1642
1643 static void lfsck_layout_unlock(struct lustre_handle *lh)
1644 {
1645         if (lustre_handle_is_used(lh)) {
1646                 ldlm_lock_decref(lh, LCK_EX);
1647                 memset(lh, 0, sizeof(*lh));
1648         }
1649 }
1650
1651 static int lfsck_layout_trans_stop(const struct lu_env *env,
1652                                    struct dt_device *dev,
1653                                    struct thandle *handle, int result)
1654 {
1655         int rc;
1656
1657         handle->th_result = result;
1658         rc = dt_trans_stop(env, dev, handle);
1659         if (rc > 0)
1660                 rc = 0;
1661         else if (rc == 0)
1662                 rc = 1;
1663
1664         return rc;
1665 }
1666
1667 /**
1668  * \retval       +1: repaired
1669  * \retval        0: did nothing
1670  * \retval      -ve: on error
1671  */
1672 static int lfsck_layout_refill_lovea(const struct lu_env *env,
1673                                      struct thandle *handle,
1674                                      struct dt_object *parent,
1675                                      struct lu_fid *cfid,
1676                                      struct lu_buf *buf,
1677                                      struct lov_ost_data_v1 *slot,
1678                                      int fl, __u32 ost_idx)
1679 {
1680         struct ost_id   *oi     = &lfsck_env_info(env)->lti_oi;
1681         int              rc;
1682
1683         fid_to_ostid(cfid, oi);
1684         ostid_cpu_to_le(oi, &slot->l_ost_oi);
1685         slot->l_ost_gen = cpu_to_le32(0);
1686         slot->l_ost_idx = cpu_to_le32(ost_idx);
1687         rc = dt_xattr_set(env, parent, buf, XATTR_NAME_LOV, fl, handle,
1688                           BYPASS_CAPA);
1689         if (rc == 0)
1690                 rc = 1;
1691
1692         return rc;
1693 }
1694
1695 /**
1696  * \retval       +1: repaired
1697  * \retval        0: did nothing
1698  * \retval      -ve: on error
1699  */
1700 static int lfsck_layout_extend_lovea(const struct lu_env *env,
1701                                      struct thandle *handle,
1702                                      struct dt_object *parent,
1703                                      struct lu_fid *cfid,
1704                                      struct lu_buf *buf, int fl,
1705                                      __u32 ost_idx, __u32 ea_off)
1706 {
1707         struct lov_mds_md_v1    *lmm    = buf->lb_buf;
1708         struct lov_ost_data_v1  *objs;
1709         int                      rc;
1710         ENTRY;
1711
1712         if (fl == LU_XATTR_CREATE) {
1713                 LASSERT(buf->lb_len == lov_mds_md_size(ea_off + 1,
1714                                                        LOV_MAGIC_V1));
1715
1716                 memset(lmm, 0, buf->lb_len);
1717                 lmm->lmm_magic = cpu_to_le32(LOV_MAGIC_V1);
1718                 /* XXX: currently, we only support LOV_PATTERN_RAID0. */
1719                 lmm->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
1720                 fid_to_lmm_oi(lfsck_dto2fid(parent), &lmm->lmm_oi);
1721                 lmm_oi_cpu_to_le(&lmm->lmm_oi, &lmm->lmm_oi);
1722                 /* XXX: We cannot know the stripe size,
1723                  *      then use the default value (1 MB). */
1724                 lmm->lmm_stripe_size = cpu_to_le32(1024 * 1024);
1725                 lmm->lmm_layout_gen = cpu_to_le16(0);
1726                 objs = &(lmm->lmm_objects[ea_off]);
1727         } else {
1728                 __u16   count = le16_to_cpu(lmm->lmm_stripe_count);
1729                 int     gap   = ea_off - count;
1730                 __u32   magic = le32_to_cpu(lmm->lmm_magic);
1731
1732                 /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3
1733                  * which has been verified in lfsck_layout_verify_header()
1734                  * already. If some new magic introduced in the future,
1735                  * then layout LFSCK needs to be updated also. */
1736                 if (magic == LOV_MAGIC_V1) {
1737                         objs = &(lmm->lmm_objects[count]);
1738                 } else {
1739                         LASSERT(magic == LOV_MAGIC_V3);
1740                         objs = &((struct lov_mds_md_v3 *)lmm)->
1741                                                         lmm_objects[count];
1742                 }
1743
1744                 if (gap > 0)
1745                         memset(objs, 0, gap * sizeof(*objs));
1746                 lmm->lmm_layout_gen =
1747                             cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
1748                 objs += gap;
1749
1750                 LASSERT(buf->lb_len == lov_mds_md_size(ea_off + 1, magic));
1751         }
1752
1753         lmm->lmm_stripe_count = cpu_to_le16(ea_off + 1);
1754         rc = lfsck_layout_refill_lovea(env, handle, parent, cfid, buf, objs,
1755                                        fl, ost_idx);
1756
1757         RETURN(rc);
1758 }
1759
1760 /**
1761  * \retval       +1: repaired
1762  * \retval        0: did nothing
1763  * \retval      -ve: on error
1764  */
1765 static int lfsck_layout_update_pfid(const struct lu_env *env,
1766                                     struct lfsck_component *com,
1767                                     struct dt_object *parent,
1768                                     struct lu_fid *cfid,
1769                                     struct dt_device *cdev, __u32 ea_off)
1770 {
1771         struct filter_fid       *pfid   = &lfsck_env_info(env)->lti_new_pfid;
1772         struct dt_object        *child;
1773         struct thandle          *handle;
1774         const struct lu_fid     *tfid   = lu_object_fid(&parent->do_lu);
1775         struct lu_buf           *buf;
1776         int                      rc     = 0;
1777         ENTRY;
1778
1779         child = lfsck_object_find_by_dev(env, cdev, cfid);
1780         if (IS_ERR(child))
1781                 RETURN(PTR_ERR(child));
1782
1783         handle = dt_trans_create(env, cdev);
1784         if (IS_ERR(handle))
1785                 GOTO(out, rc = PTR_ERR(handle));
1786
1787         pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
1788         pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
1789         /* In fact, the ff_parent::f_ver is not the real parent FID::f_ver,
1790          * instead, it is the OST-object index in its parent MDT-object
1791          * layout EA. */
1792         pfid->ff_parent.f_ver = cpu_to_le32(ea_off);
1793         buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
1794
1795         rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
1796         if (rc != 0)
1797                 GOTO(stop, rc);
1798
1799         rc = dt_trans_start(env, cdev, handle);
1800         if (rc != 0)
1801                 GOTO(stop, rc);
1802
1803         rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle,
1804                           BYPASS_CAPA);
1805
1806         GOTO(stop, rc = (rc == 0 ? 1 : rc));
1807
1808 stop:
1809         dt_trans_stop(env, cdev, handle);
1810
1811 out:
1812         lu_object_put(env, &child->do_lu);
1813
1814         return rc;
1815 }
1816
1817 /**
1818  * \retval       +1: repaired
1819  * \retval        0: did nothing
1820  * \retval      -ve: on error
1821  */
1822 static int lfsck_layout_recreate_parent(const struct lu_env *env,
1823                                         struct lfsck_component *com,
1824                                         struct lfsck_tgt_desc *ltd,
1825                                         struct lu_orphan_rec *rec,
1826                                         struct lu_fid *cfid,
1827                                         const char *prefix,
1828                                         const char *postfix,
1829                                         __u32 ea_off)
1830 {
1831         struct lfsck_thread_info        *info   = lfsck_env_info(env);
1832         char                            *name   = info->lti_key;
1833         struct lu_attr                  *la     = &info->lti_la;
1834         struct dt_object_format         *dof    = &info->lti_dof;
1835         struct lfsck_instance           *lfsck  = com->lc_lfsck;
1836         struct lu_fid                   *pfid   = &rec->lor_fid;
1837         struct lu_fid                   *tfid   = &info->lti_fid3;
1838         struct dt_device                *next   = lfsck->li_next;
1839         struct dt_object                *pobj   = NULL;
1840         struct dt_object                *cobj   = NULL;
1841         struct thandle                  *th     = NULL;
1842         struct lu_buf                   *pbuf   = NULL;
1843         struct lu_buf                   *ea_buf = &info->lti_big_buf;
1844         int                              buflen = ea_buf->lb_len;
1845         int                              rc     = 0;
1846         ENTRY;
1847
1848         /* Create .lustre/lost+found/MDTxxxx when needed. */
1849         if (unlikely(lfsck->li_lpf_obj == NULL)) {
1850                 rc = lfsck_create_lpf(env, lfsck);
1851                 if (rc != 0)
1852                         RETURN(rc);
1853         }
1854
1855         if (fid_is_zero(pfid)) {
1856                 struct filter_fid *ff = &info->lti_new_pfid;
1857
1858                 rc = lfsck_fid_alloc(env, lfsck, pfid, false);
1859                 if (rc != 0)
1860                         RETURN(rc);
1861
1862                 ff->ff_parent.f_seq = cpu_to_le64(pfid->f_seq);
1863                 ff->ff_parent.f_oid = cpu_to_le32(pfid->f_oid);
1864                 /* In fact, the ff_parent::f_ver is not the real parent FID::f_ver,
1865                  * instead, it is the OST-object index in its parent MDT-object
1866                  * layout EA. */
1867                 ff->ff_parent.f_ver = cpu_to_le32(ea_off);
1868                 pbuf = lfsck_buf_get(env, ff, sizeof(struct filter_fid));
1869                 cobj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, cfid);
1870                 if (IS_ERR(cobj))
1871                         RETURN(PTR_ERR(cobj));
1872         }
1873
1874         CDEBUG(D_LFSCK, "Re-create the lost MDT-object: parent "
1875                DFID", child "DFID", OST-index %u, stripe-index %u, "
1876                "prefix %s, postfix %s\n",
1877                PFID(pfid), PFID(cfid), ltd->ltd_index, ea_off, prefix, postfix);
1878
1879         pobj = lfsck_object_find_by_dev(env, lfsck->li_bottom, pfid);
1880         if (IS_ERR(pobj))
1881                 GOTO(put, rc = PTR_ERR(pobj));
1882
1883         LASSERT(prefix != NULL);
1884         LASSERT(postfix != NULL);
1885
1886         /** name rules:
1887          *
1888          *  1. Use the MDT-object's FID as the name with prefix and postfix.
1889          *
1890          *  1.1 prefix "C-":    More than one OST-objects cliam the same
1891          *                      MDT-object and the same slot in the layout EA.
1892          *                      It may be created for dangling referenced MDT
1893          *                      object or may be not.
1894          *  1.2 prefix "N-":    The orphan OST-object does not know which one
1895          *                      is the real parent, so the LFSCK assign a new
1896          *                      FID as its parent.
1897          *  1.3 prefix "R-":    The orphan OST-object know its parent FID but
1898          *                      does not know the position in the namespace.
1899          *
1900          *  2. If there is name conflict, increase FID::f_ver for new name. */
1901         sprintf(name, "%s"DFID"%s", prefix, PFID(pfid), postfix);
1902         do {
1903                 rc = dt_lookup(env, lfsck->li_lpf_obj, (struct dt_rec *)tfid,
1904                                (const struct dt_key *)name, BYPASS_CAPA);
1905                 if (rc != 0 && rc != -ENOENT)
1906                         GOTO(put, rc);
1907
1908                 if (unlikely(rc == 0)) {
1909                         CWARN("%s: The name %s under lost+found has been used "
1910                               "by the "DFID". Try to increase the FID version "
1911                               "for the new file name.\n",
1912                               lfsck_lfsck2name(lfsck), name, PFID(tfid));
1913                         *tfid = *pfid;
1914                         tfid->f_ver++;
1915                         sprintf(name, "%s"DFID"%s", prefix, PFID(tfid), postfix);
1916                 }
1917         } while (rc == 0);
1918
1919         memset(la, 0, sizeof(*la));
1920         la->la_uid = rec->lor_uid;
1921         la->la_gid = rec->lor_gid;
1922         la->la_mode = S_IFREG | S_IRUSR | S_IWUSR;
1923         la->la_valid = LA_MODE | LA_UID | LA_GID;
1924
1925         memset(dof, 0, sizeof(*dof));
1926         dof->dof_type = dt_mode_to_dft(S_IFREG);
1927
1928         rc = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
1929         if (buflen < rc) {
1930                 lu_buf_realloc(ea_buf, rc);
1931                 buflen = ea_buf->lb_len;
1932                 if (ea_buf->lb_buf == NULL)
1933                         GOTO(put, rc = -ENOMEM);
1934         } else {
1935                 ea_buf->lb_len = rc;
1936         }
1937
1938         th = dt_trans_create(env, next);
1939         if (IS_ERR(th))
1940                 GOTO(put, rc = PTR_ERR(th));
1941
1942         /* 1a. Update OST-object's parent information remotely.
1943          *
1944          * If other subsequent modifications failed, then next LFSCK scanning
1945          * will process the OST-object as orphan again with known parent FID. */
1946         if (cobj != NULL) {
1947                 rc = dt_declare_xattr_set(env, cobj, pbuf, XATTR_NAME_FID, 0, th);
1948                 if (rc != 0)
1949                         GOTO(stop, rc);
1950         }
1951
1952         /* 2a. Create the MDT-object locally. */
1953         rc = dt_declare_create(env, pobj, la, NULL, dof, th);
1954         if (rc != 0)
1955                 GOTO(stop, rc);
1956
1957         /* 3a. Add layout EA for the MDT-object. */
1958         rc = dt_declare_xattr_set(env, pobj, ea_buf, XATTR_NAME_LOV,
1959                                   LU_XATTR_CREATE, th);
1960         if (rc != 0)
1961                 GOTO(stop, rc);
1962
1963         /* 4a. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */
1964         rc = dt_declare_insert(env, lfsck->li_lpf_obj,
1965                                (const struct dt_rec *)pfid,
1966                                (const struct dt_key *)name, th);
1967         if (rc != 0)
1968                 GOTO(stop, rc);
1969
1970         rc = dt_trans_start(env, next, th);
1971         if (rc != 0)
1972                 GOTO(stop, rc);
1973
1974         /* 1b. Update OST-object's parent information remotely. */
1975         if (cobj != NULL) {
1976                 rc = dt_xattr_set(env, cobj, pbuf, XATTR_NAME_FID, 0, th,
1977                                   BYPASS_CAPA);
1978                 if (rc != 0)
1979                         GOTO(stop, rc);
1980         }
1981
1982         dt_write_lock(env, pobj, 0);
1983         /* 2b. Create the MDT-object locally. */
1984         rc = dt_create(env, pobj, la, NULL, dof, th);
1985         if (rc == 0)
1986                 /* 3b. Add layout EA for the MDT-object. */
1987                 rc = lfsck_layout_extend_lovea(env, th, pobj, cfid, ea_buf,
1988                                                LU_XATTR_CREATE, ltd->ltd_index,
1989                                                ea_off);
1990         dt_write_unlock(env, pobj);
1991         if (rc < 0)
1992                 GOTO(stop, rc);
1993
1994         /* 4b. Insert the MDT-object to .lustre/lost+found/MDTxxxx/ */
1995         rc = dt_insert(env, lfsck->li_lpf_obj,
1996                        (const struct dt_rec *)pfid,
1997                        (const struct dt_key *)name, th, BYPASS_CAPA, 1);
1998
1999         GOTO(stop, rc);
2000
2001 stop:
2002         dt_trans_stop(env, next, th);
2003 put:
2004         if (cobj != NULL && !IS_ERR(cobj))
2005                 lu_object_put(env, &cobj->do_lu);
2006         if (pobj != NULL && !IS_ERR(pobj))
2007                 lu_object_put(env, &pobj->do_lu);
2008         ea_buf->lb_len = buflen;
2009
2010         return rc >= 0 ? 1 : rc;
2011 }
2012
2013 static int lfsck_layout_master_conditional_destroy(const struct lu_env *env,
2014                                                    struct lfsck_component *com,
2015                                                    const struct lu_fid *fid,
2016                                                    __u32 index)
2017 {
2018         struct lfsck_thread_info *info  = lfsck_env_info(env);
2019         struct lfsck_request     *lr    = &info->lti_lr;
2020         struct lfsck_instance    *lfsck = com->lc_lfsck;
2021         struct lfsck_tgt_desc    *ltd;
2022         struct ptlrpc_request    *req;
2023         struct lfsck_request     *tmp;
2024         struct obd_export        *exp;
2025         int                       rc    = 0;
2026         ENTRY;
2027
2028         ltd = lfsck_tgt_get(&lfsck->li_ost_descs, index);
2029         if (unlikely(ltd == NULL))
2030                 RETURN(-ENODEV);
2031
2032         exp = ltd->ltd_exp;
2033         if (!(exp_connect_flags(exp) & OBD_CONNECT_LFSCK))
2034                 GOTO(put, rc = -EOPNOTSUPP);
2035
2036         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
2037         if (req == NULL)
2038                 GOTO(put, rc = -ENOMEM);
2039
2040         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
2041         if (rc != 0) {
2042                 ptlrpc_request_free(req);
2043
2044                 GOTO(put, rc);
2045         }
2046
2047         memset(lr, 0, sizeof(*lr));
2048         lr->lr_event = LE_CONDITIONAL_DESTROY;
2049         lr->lr_active = LT_LAYOUT;
2050         lr->lr_fid = *fid;
2051
2052         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
2053         *tmp = *lr;
2054         ptlrpc_request_set_replen(req);
2055
2056         rc = ptlrpc_queue_wait(req);
2057         ptlrpc_req_finished(req);
2058
2059         GOTO(put, rc);
2060
2061 put:
2062         lfsck_tgt_put(ltd);
2063
2064         return rc;
2065 }
2066
2067 static int lfsck_layout_slave_conditional_destroy(const struct lu_env *env,
2068                                                   struct lfsck_component *com,
2069                                                   struct lfsck_request *lr)
2070 {
2071         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2072         struct lu_attr                  *la     = &info->lti_la;
2073         ldlm_policy_data_t              *policy = &info->lti_policy;
2074         struct ldlm_res_id              *resid  = &info->lti_resid;
2075         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2076         struct dt_device                *dev    = lfsck->li_bottom;
2077         struct lu_fid                   *fid    = &lr->lr_fid;
2078         struct dt_object                *obj;
2079         struct thandle                  *th     = NULL;
2080         struct lustre_handle             lh     = { 0 };
2081         __u64                            flags  = 0;
2082         int                              rc     = 0;
2083         ENTRY;
2084
2085         obj = lfsck_object_find_by_dev(env, dev, fid);
2086         if (IS_ERR(obj))
2087                 RETURN(PTR_ERR(obj));
2088
2089         dt_read_lock(env, obj, 0);
2090         if (dt_object_exists(obj) == 0) {
2091                 dt_read_unlock(env, obj);
2092
2093                 GOTO(put, rc = -ENOENT);
2094         }
2095
2096         /* Get obj's attr without lock firstly. */
2097         rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
2098         dt_read_unlock(env, obj);
2099         if (rc != 0)
2100                 GOTO(put, rc);
2101
2102         if (likely(la->la_ctime != 0 || la->la_mode & S_ISUID))
2103                 GOTO(put, rc = -ETXTBSY);
2104
2105         /* Acquire extent lock on [0, EOF] to sync with all possible written. */
2106         LASSERT(lfsck->li_namespace != NULL);
2107
2108         memset(policy, 0, sizeof(*policy));
2109         policy->l_extent.end = OBD_OBJECT_EOF;
2110         ost_fid_build_resid(fid, resid);
2111         rc = ldlm_cli_enqueue_local(lfsck->li_namespace, resid, LDLM_EXTENT,
2112                                     policy, LCK_EX, &flags, ldlm_blocking_ast,
2113                                     ldlm_completion_ast, NULL, NULL, 0,
2114                                     LVB_T_NONE, NULL, &lh);
2115         if (rc != ELDLM_OK)
2116                 GOTO(put, rc = -EIO);
2117
2118         th = dt_trans_create(env, dev);
2119         if (IS_ERR(th))
2120                 GOTO(unlock1, rc = PTR_ERR(th));
2121
2122         rc = dt_declare_ref_del(env, obj, th);
2123         if (rc != 0)
2124                 GOTO(stop, rc);
2125
2126         rc = dt_declare_destroy(env, obj, th);
2127         if (rc != 0)
2128                 GOTO(stop, rc);
2129
2130         rc = dt_trans_start_local(env, dev, th);
2131         if (rc != 0)
2132                 GOTO(stop, rc);
2133
2134         dt_write_lock(env, obj, 0);
2135         /* Get obj's attr within lock again. */
2136         rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
2137         if (rc != 0)
2138                 GOTO(unlock2, rc);
2139
2140         if (la->la_ctime != 0)
2141                 GOTO(unlock2, rc = -ETXTBSY);
2142
2143         rc = dt_ref_del(env, obj, th);
2144         if (rc != 0)
2145                 GOTO(unlock2, rc);
2146
2147         rc = dt_destroy(env, obj, th);
2148         if (rc == 0)
2149                 CDEBUG(D_LFSCK, "Destroy the empty OST-object "DFID" which "
2150                        "was created for reparing dangling referenced case. "
2151                        "But the original missed OST-object is found now.\n",
2152                        PFID(fid));
2153
2154         GOTO(unlock2, rc);
2155
2156 unlock2:
2157         dt_write_unlock(env, obj);
2158
2159 stop:
2160         dt_trans_stop(env, dev, th);
2161
2162 unlock1:
2163         ldlm_lock_decref(&lh, LCK_EX);
2164
2165 put:
2166         lu_object_put(env, &obj->do_lu);
2167
2168         return rc;
2169 }
2170
2171 /**
2172  * Some OST-object has occupied the specified layout EA slot.
2173  * Such OST-object may be generated by the LFSCK when repair
2174  * dangling referenced MDT-object, which can be indicated by
2175  * attr::la_ctime == 0 but without S_ISUID in la_mode. If it
2176  * is true and such OST-object has not been modified yet, we
2177  * will replace it with the orphan OST-object; otherwise the
2178  * LFSCK will create new MDT-object to reference the orphan.
2179  *
2180  * \retval       +1: repaired
2181  * \retval        0: did nothing
2182  * \retval      -ve: on error
2183  */
2184 static int lfsck_layout_conflict_create(const struct lu_env *env,
2185                                         struct lfsck_component *com,
2186                                         struct lfsck_tgt_desc *ltd,
2187                                         struct lu_orphan_rec *rec,
2188                                         struct dt_object *parent,
2189                                         struct lu_fid *cfid,
2190                                         struct lu_buf *ea_buf,
2191                                         struct lov_ost_data_v1 *slot,
2192                                         __u32 ea_off, __u32 ori_len)
2193 {
2194         struct lfsck_thread_info *info          = lfsck_env_info(env);
2195         struct lu_fid            *cfid2         = &info->lti_fid2;
2196         struct ost_id            *oi            = &info->lti_oi;
2197         struct lov_mds_md_v1     *lmm           = ea_buf->lb_buf;
2198         struct dt_device         *dev           = com->lc_lfsck->li_bottom;
2199         struct thandle           *th            = NULL;
2200         struct lustre_handle      lh            = { 0 };
2201         char                      postfix[64];
2202         __u32                     ost_idx2      = le32_to_cpu(slot->l_ost_idx);
2203         int                       rc            = 0;
2204         ENTRY;
2205
2206         ostid_le_to_cpu(&slot->l_ost_oi, oi);
2207         ostid_to_fid(cfid2, oi, ost_idx2);
2208
2209         CDEBUG(D_LFSCK, "Handle layout EA conflict: parent "DFID
2210                ", cur-child "DFID" on the OST %u, orphan-child "
2211                DFID" on the OST %u, stripe-index %u\n",
2212                PFID(lfsck_dto2fid(parent)), PFID(cfid2), ost_idx2,
2213                PFID(cfid), ltd->ltd_index, ea_off);
2214
2215         /* Hold layout lock on the parent to prevent others to access. */
2216         rc = lfsck_layout_lock(env, com, parent, &lh,
2217                                MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR);
2218         if (rc != 0)
2219                 GOTO(out, rc);
2220
2221         rc = lfsck_layout_master_conditional_destroy(env, com, cfid2, ost_idx2);
2222
2223         /* If the conflict OST-obejct is not created for fixing dangling
2224          * referenced MDT-object in former LFSCK check/repair, or it has
2225          * been modified by others, then we cannot destroy it. Re-create
2226          * a new MDT-object for the orphan OST-object. */
2227         if (rc == -ETXTBSY) {
2228                 /* No need the layout lock on the original parent. */
2229                 lfsck_layout_unlock(&lh);
2230                 ea_buf->lb_len = ori_len;
2231
2232                 fid_zero(&rec->lor_fid);
2233                 snprintf(postfix, 64, "-"DFID"-%x",
2234                          PFID(lu_object_fid(&parent->do_lu)), ea_off);
2235                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2236                                                   "C-", postfix, ea_off);
2237
2238                 RETURN(rc);
2239         }
2240
2241         if (rc != 0 && rc != -ENOENT)
2242                 GOTO(unlock, rc);
2243
2244         th = dt_trans_create(env, dev);
2245         if (IS_ERR(th))
2246                 GOTO(unlock, rc = PTR_ERR(th));
2247
2248         rc = dt_declare_xattr_set(env, parent, ea_buf, XATTR_NAME_LOV,
2249                                   LU_XATTR_REPLACE, th);
2250         if (rc != 0)
2251                 GOTO(stop, rc);
2252
2253         rc = dt_trans_start_local(env, dev, th);
2254         if (rc != 0)
2255                 GOTO(stop, rc);
2256
2257         dt_write_lock(env, parent, 0);
2258         lmm->lmm_layout_gen = cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2259         rc = lfsck_layout_refill_lovea(env, th, parent, cfid, ea_buf, slot,
2260                                        LU_XATTR_REPLACE, ltd->ltd_index);
2261         dt_write_unlock(env, parent);
2262
2263         GOTO(stop, rc);
2264
2265 stop:
2266         dt_trans_stop(env, dev, th);
2267
2268 unlock:
2269         lfsck_layout_unlock(&lh);
2270
2271 out:
2272         ea_buf->lb_len = ori_len;
2273
2274         return rc >= 0 ? 1 : rc;
2275 }
2276
2277 /**
2278  * \retval       +1: repaired
2279  * \retval        0: did nothing
2280  * \retval      -ve: on error
2281  */
2282 static int lfsck_layout_recreate_lovea(const struct lu_env *env,
2283                                        struct lfsck_component *com,
2284                                        struct lfsck_tgt_desc *ltd,
2285                                        struct lu_orphan_rec *rec,
2286                                        struct dt_object *parent,
2287                                        struct lu_fid *cfid,
2288                                        __u32 ost_idx, __u32 ea_off)
2289 {
2290         struct lfsck_thread_info *info          = lfsck_env_info(env);
2291         struct lu_buf            *buf           = &info->lti_big_buf;
2292         struct lu_fid            *fid           = &info->lti_fid2;
2293         struct ost_id            *oi            = &info->lti_oi;
2294         struct lfsck_instance    *lfsck         = com->lc_lfsck;
2295         struct dt_device         *dt            = lfsck->li_bottom;
2296         struct lfsck_bookmark    *bk            = &lfsck->li_bookmark_ram;
2297         struct thandle            *handle       = NULL;
2298         size_t                    buflen        = buf->lb_len;
2299         struct lov_mds_md_v1     *lmm;
2300         struct lov_ost_data_v1   *objs;
2301         struct lustre_handle      lh            = { 0 };
2302         __u32                     magic;
2303         int                       fl            = 0;
2304         int                       rc;
2305         int                       rc1;
2306         int                       i;
2307         __u16                     count;
2308         ENTRY;
2309
2310         CDEBUG(D_LFSCK, "Re-create the crashed layout EA: parent "
2311                DFID", child "DFID", OST-index %u, stripe-index %u\n",
2312                PFID(lfsck_dto2fid(parent)), PFID(cfid), ost_idx, ea_off);
2313
2314         rc = lfsck_layout_lock(env, com, parent, &lh,
2315                                MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR);
2316         if (rc != 0)
2317                 RETURN(rc);
2318
2319 again:
2320         if (!(bk->lb_param & LPF_DRYRUN)) {
2321                 handle = dt_trans_create(env, dt);
2322                 if (IS_ERR(handle))
2323                         GOTO(unlock_layout, rc = PTR_ERR(handle));
2324
2325                 rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2326                                           fl, handle);
2327                 if (rc != 0)
2328                         GOTO(stop, rc);
2329
2330                 rc = dt_trans_start_local(env, dt, handle);
2331                 if (rc != 0)
2332                         GOTO(stop, rc);
2333         }
2334
2335         dt_write_lock(env, parent, 0);
2336         rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV, BYPASS_CAPA);
2337         if (rc == -ERANGE) {
2338                 rc = dt_xattr_get(env, parent, &LU_BUF_NULL, XATTR_NAME_LOV,
2339                                   BYPASS_CAPA);
2340                 LASSERT(rc != 0);
2341
2342                 dt_write_unlock(env, parent);
2343                 if (handle != NULL) {
2344                         dt_trans_stop(env, dt, handle);
2345                         handle = NULL;
2346                 }
2347
2348                 if (rc < 0)
2349                         GOTO(unlock_layout, rc);
2350
2351                 lu_buf_realloc(buf, rc);
2352                 buflen = buf->lb_len;
2353                 if (buf->lb_buf == NULL)
2354                         GOTO(unlock_layout, rc = -ENOMEM);
2355
2356                 fl = LU_XATTR_REPLACE;
2357                 goto again;
2358         } else if (rc == -ENODATA || rc == 0) {
2359                 fl = LU_XATTR_CREATE;
2360         } else if (rc < 0) {
2361                 GOTO(unlock_parent, rc);
2362         } else if (unlikely(buf->lb_len == 0)) {
2363                 dt_write_unlock(env, parent);
2364                 if (handle != NULL) {
2365                         dt_trans_stop(env, dt, handle);
2366                         handle = NULL;
2367                 }
2368
2369                 lu_buf_alloc(buf, rc);
2370                 buflen = buf->lb_len;
2371                 if (buf->lb_buf == NULL)
2372                         GOTO(unlock_layout, rc = -ENOMEM);
2373
2374                 fl = LU_XATTR_REPLACE;
2375                 goto again;
2376         } else {
2377                 fl = LU_XATTR_REPLACE;
2378         }
2379
2380         if (fl == LU_XATTR_CREATE) {
2381                 if (bk->lb_param & LPF_DRYRUN)
2382                         GOTO(unlock_parent, rc = 1);
2383
2384                 rc = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
2385                 /* If the declared is not big enough, re-try. */
2386                 if (buf->lb_len < rc) {
2387                         dt_write_unlock(env, parent);
2388                         if (handle != NULL) {
2389                                 dt_trans_stop(env, dt, handle);
2390                                 handle = NULL;
2391                         }
2392
2393                         lu_buf_realloc(buf, rc);
2394                         buflen = buf->lb_len;
2395                         if (buf->lb_buf == NULL)
2396                                 GOTO(unlock_layout, rc = -ENOMEM);
2397
2398                         goto again;
2399                 }
2400
2401                 buf->lb_len = rc;
2402                 rc = lfsck_layout_extend_lovea(env, handle, parent, cfid, buf,
2403                                                fl, ost_idx, ea_off);
2404
2405                 GOTO(unlock_parent, rc);
2406         }
2407
2408         lmm = buf->lb_buf;
2409         rc1 = lfsck_layout_verify_header(lmm);
2410         if (rc1 != 0)
2411                 GOTO(unlock_parent, rc = rc1);
2412
2413         /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2414          * been verified in lfsck_layout_verify_header() already. If some
2415          * new magic introduced in the future, then layout LFSCK needs to
2416          * be updated also. */
2417         magic = le32_to_cpu(lmm->lmm_magic);
2418         if (magic == LOV_MAGIC_V1) {
2419                 objs = &(lmm->lmm_objects[0]);
2420         } else {
2421                 LASSERT(magic == LOV_MAGIC_V3);
2422                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
2423         }
2424
2425         count = le16_to_cpu(lmm->lmm_stripe_count);
2426         if (count == 0)
2427                 GOTO(unlock_parent, rc = -EINVAL);
2428         LASSERT(count > 0);
2429
2430         /* Exceed the current end of MDT-object layout EA. Then extend it. */
2431         if (count <= ea_off) {
2432                 if (bk->lb_param & LPF_DRYRUN)
2433                         GOTO(unlock_parent, rc = 1);
2434
2435                 rc = lov_mds_md_size(ea_off + 1, LOV_MAGIC_V1);
2436                 /* If the declared is not big enough, re-try. */
2437                 if (buf->lb_len < rc) {
2438                         dt_write_unlock(env, parent);
2439                         if (handle != NULL) {
2440                                 dt_trans_stop(env, dt, handle);
2441                                 handle = NULL;
2442                         }
2443
2444                         lu_buf_realloc(buf, rc);
2445                         buflen = buf->lb_len;
2446                         if (buf->lb_buf == NULL)
2447                                 GOTO(unlock_layout, rc = -ENOMEM);
2448
2449                         goto again;
2450                 }
2451
2452                 buf->lb_len = rc;
2453                 rc = lfsck_layout_extend_lovea(env, handle, parent, cfid, buf,
2454                                                fl, ost_idx, ea_off);
2455                 GOTO(unlock_parent, rc);
2456         }
2457
2458         LASSERTF(rc > 0, "invalid rc = %d\n", rc);
2459
2460         buf->lb_len = rc;
2461         for (i = 0; i < count; i++, objs++) {
2462                 /* The MDT-object was created via lfsck_layout_recover_create()
2463                  * by others before, and we fill the dummy layout EA. */
2464                 if (is_dummy_lov_ost_data(objs)) {
2465                         if (i != ea_off)
2466                                 continue;
2467
2468                         if (bk->lb_param & LPF_DRYRUN)
2469                                 GOTO(unlock_parent, rc = 1);
2470
2471                         lmm->lmm_layout_gen =
2472                             cpu_to_le16(le16_to_cpu(lmm->lmm_layout_gen) + 1);
2473                         rc = lfsck_layout_refill_lovea(env, handle, parent,
2474                                                        cfid, buf, objs, fl,
2475                                                        ost_idx);
2476                         GOTO(unlock_parent, rc);
2477                 }
2478
2479                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
2480                 ostid_to_fid(fid, oi, le32_to_cpu(objs->l_ost_idx));
2481                 /* It should be rare case, the slot is there, but the LFSCK
2482                  * does not handle it during the first-phase cycle scanning. */
2483                 if (unlikely(lu_fid_eq(fid, cfid))) {
2484                         if (i == ea_off) {
2485                                 GOTO(unlock_parent, rc = 0);
2486                         } else {
2487                                 /* Rare case that the OST-object index
2488                                  * does not match the parent MDT-object
2489                                  * layout EA. We trust the later one. */
2490                                 if (bk->lb_param & LPF_DRYRUN)
2491                                         GOTO(unlock_parent, rc = 1);
2492
2493                                 dt_write_unlock(env, parent);
2494                                 if (handle != NULL)
2495                                         dt_trans_stop(env, dt, handle);
2496                                 lfsck_layout_unlock(&lh);
2497                                 buf->lb_len = buflen;
2498                                 rc = lfsck_layout_update_pfid(env, com, parent,
2499                                                         cfid, ltd->ltd_tgt, i);
2500
2501                                 RETURN(rc);
2502                         }
2503                 }
2504         }
2505
2506         /* The MDT-object exists, but related layout EA slot is occupied
2507          * by others. */
2508         if (bk->lb_param & LPF_DRYRUN)
2509                 GOTO(unlock_parent, rc = 1);
2510
2511         dt_write_unlock(env, parent);
2512         if (handle != NULL)
2513                 dt_trans_stop(env, dt, handle);
2514         lfsck_layout_unlock(&lh);
2515         if (le32_to_cpu(lmm->lmm_magic) == LOV_MAGIC_V1)
2516                 objs = &(lmm->lmm_objects[ea_off]);
2517         else
2518                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[ea_off];
2519         rc = lfsck_layout_conflict_create(env, com, ltd, rec, parent, cfid,
2520                                           buf, objs, ea_off, buflen);
2521
2522         RETURN(rc);
2523
2524 unlock_parent:
2525         dt_write_unlock(env, parent);
2526
2527 stop:
2528         if (handle != NULL)
2529                 dt_trans_stop(env, dt, handle);
2530
2531 unlock_layout:
2532         lfsck_layout_unlock(&lh);
2533         buf->lb_len = buflen;
2534
2535         return rc;
2536 }
2537
2538 static int lfsck_layout_scan_orphan_one(const struct lu_env *env,
2539                                         struct lfsck_component *com,
2540                                         struct lfsck_tgt_desc *ltd,
2541                                         struct lu_orphan_rec *rec,
2542                                         struct lu_fid *cfid)
2543 {
2544         struct lfsck_layout     *lo     = com->lc_file_ram;
2545         struct lu_fid           *pfid   = &rec->lor_fid;
2546         struct dt_object        *parent = NULL;
2547         __u32                    ea_off = pfid->f_ver;
2548         int                      rc     = 0;
2549         ENTRY;
2550
2551         if (!fid_is_sane(cfid))
2552                 GOTO(out, rc = -EINVAL);
2553
2554         if (fid_is_zero(pfid)) {
2555                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2556                                                   "N-", "", ea_off);
2557                 GOTO(out, rc);
2558         }
2559
2560         pfid->f_ver = 0;
2561         if (!fid_is_sane(pfid))
2562                 GOTO(out, rc = -EINVAL);
2563
2564         parent = lfsck_object_find_by_dev(env, com->lc_lfsck->li_bottom, pfid);
2565         if (IS_ERR(parent))
2566                 GOTO(out, rc = PTR_ERR(parent));
2567
2568         if (unlikely(dt_object_remote(parent) != 0))
2569                 GOTO(put, rc = -EXDEV);
2570
2571         if (dt_object_exists(parent) == 0) {
2572                 lu_object_put(env, &parent->do_lu);
2573                 rc = lfsck_layout_recreate_parent(env, com, ltd, rec, cfid,
2574                                                   "R-", "", ea_off);
2575                 GOTO(out, rc);
2576         }
2577
2578         if (!S_ISREG(lu_object_attr(&parent->do_lu)))
2579                 GOTO(put, rc = -EISDIR);
2580
2581         rc = lfsck_layout_recreate_lovea(env, com, ltd, rec, parent, cfid,
2582                                          ltd->ltd_index, ea_off);
2583
2584         GOTO(put, rc);
2585
2586 put:
2587         if (rc <= 0)
2588                 lu_object_put(env, &parent->do_lu);
2589         else
2590                 /* The layout EA is changed, need to be reloaded next time. */
2591                 lu_object_put_nocache(env, &parent->do_lu);
2592
2593 out:
2594         down_write(&com->lc_sem);
2595         com->lc_new_scanned++;
2596         com->lc_new_checked++;
2597         if (rc > 0) {
2598                 lo->ll_objs_repaired[LLIT_ORPHAN - 1]++;
2599                 rc = 0;
2600         } else if (rc < 0) {
2601                 lo->ll_objs_failed_phase2++;
2602         }
2603         up_write(&com->lc_sem);
2604
2605         return rc;
2606 }
2607
2608 static int lfsck_layout_scan_orphan(const struct lu_env *env,
2609                                     struct lfsck_component *com,
2610                                     struct lfsck_tgt_desc *ltd)
2611 {
2612         struct lfsck_layout             *lo     = com->lc_file_ram;
2613         struct lfsck_instance           *lfsck  = com->lc_lfsck;
2614         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
2615         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2616         struct ost_id                   *oi     = &info->lti_oi;
2617         struct lu_fid                   *fid    = &info->lti_fid;
2618         struct dt_object                *obj;
2619         const struct dt_it_ops          *iops;
2620         struct dt_it                    *di;
2621         int                              rc     = 0;
2622         ENTRY;
2623
2624         CDEBUG(D_LFSCK, "%s: start the orphan scanning for OST%04x\n",
2625                lfsck_lfsck2name(lfsck), ltd->ltd_index);
2626
2627         ostid_set_seq(oi, FID_SEQ_IDIF);
2628         ostid_set_id(oi, 0);
2629         ostid_to_fid(fid, oi, ltd->ltd_index);
2630         obj = lfsck_object_find_by_dev(env, ltd->ltd_tgt, fid);
2631         if (unlikely(IS_ERR(obj)))
2632                 RETURN(PTR_ERR(obj));
2633
2634         rc = obj->do_ops->do_index_try(env, obj, &dt_lfsck_orphan_features);
2635         if (rc != 0)
2636                 GOTO(put, rc);
2637
2638         iops = &obj->do_index_ops->dio_it;
2639         di = iops->init(env, obj, 0, BYPASS_CAPA);
2640         if (IS_ERR(di))
2641                 GOTO(put, rc = PTR_ERR(di));
2642
2643         rc = iops->load(env, di, 0);
2644         if (rc == -ESRCH) {
2645                 /* -ESRCH means that the orphan OST-objects rbtree has been
2646                  * cleanup because of the OSS server restart or other errors. */
2647                 lo->ll_flags |= LF_INCOMPLETE;
2648                 GOTO(fini, rc);
2649         }
2650
2651         if (rc == 0)
2652                 rc = iops->next(env, di);
2653         else if (rc > 0)
2654                 rc = 0;
2655
2656         if (rc < 0)
2657                 GOTO(fini, rc);
2658
2659         if (rc > 0)
2660                 GOTO(fini, rc = 0);
2661
2662         do {
2663                 struct dt_key           *key;
2664                 struct lu_orphan_rec    *rec = &info->lti_rec;
2665
2666                 if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_DELAY3) &&
2667                     cfs_fail_val > 0) {
2668                         struct ptlrpc_thread    *thread = &lfsck->li_thread;
2669                         struct l_wait_info       lwi;
2670
2671                         lwi = LWI_TIMEOUT(cfs_time_seconds(cfs_fail_val),
2672                                           NULL, NULL);
2673                         l_wait_event(thread->t_ctl_waitq,
2674                                      !thread_is_running(thread),
2675                                      &lwi);
2676                 }
2677
2678                 key = iops->key(env, di);
2679                 com->lc_fid_latest_scanned_phase2 = *(struct lu_fid *)key;
2680                 rc = iops->rec(env, di, (struct dt_rec *)rec, 0);
2681                 if (rc == 0)
2682                         rc = lfsck_layout_scan_orphan_one(env, com, ltd, rec,
2683                                         &com->lc_fid_latest_scanned_phase2);
2684                 if (rc != 0 && bk->lb_param & LPF_FAILOUT)
2685                         GOTO(fini, rc);
2686
2687                 lfsck_control_speed_by_self(com);
2688                 do {
2689                         rc = iops->next(env, di);
2690                 } while (rc < 0 && !(bk->lb_param & LPF_FAILOUT));
2691         } while (rc == 0);
2692
2693         GOTO(fini, rc);
2694
2695 fini:
2696         iops->put(env, di);
2697         iops->fini(env, di);
2698 put:
2699         lu_object_put(env, &obj->do_lu);
2700
2701         CDEBUG(D_LFSCK, "%s: finish the orphan scanning for OST%04x, rc = %d\n",
2702                lfsck_lfsck2name(lfsck), ltd->ltd_index, rc);
2703
2704         return rc > 0 ? 0 : rc;
2705 }
2706
2707 /* For the MDT-object with dangling reference, we need to re-create
2708  * the missed OST-object with the known FID/owner information. */
2709 static int lfsck_layout_recreate_ostobj(const struct lu_env *env,
2710                                         struct lfsck_component *com,
2711                                         struct lfsck_layout_req *llr,
2712                                         struct lu_attr *la)
2713 {
2714         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2715         struct filter_fid               *pfid   = &info->lti_new_pfid;
2716         struct dt_allocation_hint       *hint   = &info->lti_hint;
2717         struct dt_object                *parent = llr->llr_parent->llo_obj;
2718         struct dt_object                *child  = llr->llr_child;
2719         struct dt_device                *dev    = lfsck_obj2dt_dev(child);
2720         const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
2721         struct thandle                  *handle;
2722         struct lu_buf                   *buf;
2723         struct lustre_handle             lh     = { 0 };
2724         int                              rc;
2725         ENTRY;
2726
2727         CDEBUG(D_LFSCK, "Repair dangling reference for: parent "DFID
2728                ", child "DFID", OST-index %u, stripe-index %u, owner %u:%u\n",
2729                PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
2730                llr->llr_ost_idx, llr->llr_lov_idx, la->la_uid, la->la_gid);
2731
2732         rc = lfsck_layout_lock(env, com, parent, &lh,
2733                                MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR);
2734         if (rc != 0)
2735                 RETURN(rc);
2736
2737         handle = dt_trans_create(env, dev);
2738         if (IS_ERR(handle))
2739                 GOTO(unlock1, rc = PTR_ERR(handle));
2740
2741         hint->dah_parent = NULL;
2742         hint->dah_mode = 0;
2743         pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2744         pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2745         pfid->ff_parent.f_ver = cpu_to_le32(llr->llr_lov_idx);
2746         buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2747
2748         rc = dt_declare_create(env, child, la, hint, NULL, handle);
2749         if (rc != 0)
2750                 GOTO(stop, rc);
2751
2752         rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID,
2753                                   LU_XATTR_CREATE, handle);
2754         if (rc != 0)
2755                 GOTO(stop, rc);
2756
2757         rc = dt_trans_start(env, dev, handle);
2758         if (rc != 0)
2759                 GOTO(stop, rc);
2760
2761         dt_read_lock(env, parent, 0);
2762         if (unlikely(lu_object_is_dying(parent->do_lu.lo_header)))
2763                 GOTO(unlock2, rc = 1);
2764
2765         rc = dt_create(env, child, la, hint, NULL, handle);
2766         if (rc != 0)
2767                 GOTO(unlock2, rc);
2768
2769         rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, LU_XATTR_CREATE,
2770                           handle, BYPASS_CAPA);
2771
2772         GOTO(unlock2, rc);
2773
2774 unlock2:
2775         dt_read_unlock(env, parent);
2776
2777 stop:
2778         rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2779
2780 unlock1:
2781         lfsck_layout_unlock(&lh);
2782
2783         return rc;
2784 }
2785
2786 /* If the OST-object does not recognize the MDT-object as its parent, and
2787  * there is no other MDT-object claims as its parent, then just trust the
2788  * given MDT-object as its parent. So update the OST-object filter_fid. */
2789 static int lfsck_layout_repair_unmatched_pair(const struct lu_env *env,
2790                                               struct lfsck_component *com,
2791                                               struct lfsck_layout_req *llr,
2792                                               const struct lu_attr *pla)
2793 {
2794         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2795         struct filter_fid               *pfid   = &info->lti_new_pfid;
2796         struct lu_attr                  *tla    = &info->lti_la3;
2797         struct dt_object                *parent = llr->llr_parent->llo_obj;
2798         struct dt_object                *child  = llr->llr_child;
2799         struct dt_device                *dev    = lfsck_obj2dt_dev(child);
2800         const struct lu_fid             *tfid   = lu_object_fid(&parent->do_lu);
2801         struct thandle                  *handle;
2802         struct lu_buf                   *buf;
2803         struct lustre_handle             lh     = { 0 };
2804         int                              rc;
2805         ENTRY;
2806
2807         CDEBUG(D_LFSCK, "Repair unmatched MDT-OST pair for: parent "DFID
2808                ", child "DFID", OST-index %u, stripe-index %u, owner %u:%u\n",
2809                PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
2810                llr->llr_ost_idx, llr->llr_lov_idx, pla->la_uid, pla->la_gid);
2811
2812         rc = lfsck_layout_lock(env, com, parent, &lh,
2813                                MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR);
2814         if (rc != 0)
2815                 RETURN(rc);
2816
2817         handle = dt_trans_create(env, dev);
2818         if (IS_ERR(handle))
2819                 GOTO(unlock1, rc = PTR_ERR(handle));
2820
2821         pfid->ff_parent.f_seq = cpu_to_le64(tfid->f_seq);
2822         pfid->ff_parent.f_oid = cpu_to_le32(tfid->f_oid);
2823         /* The ff_parent->f_ver is not the real parent fid->f_ver. Instead,
2824          * it is the OST-object index in the parent MDT-object layout. */
2825         pfid->ff_parent.f_ver = cpu_to_le32(llr->llr_lov_idx);
2826         buf = lfsck_buf_get(env, pfid, sizeof(struct filter_fid));
2827
2828         rc = dt_declare_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle);
2829         if (rc != 0)
2830                 GOTO(stop, rc);
2831
2832         tla->la_valid = LA_UID | LA_GID;
2833         tla->la_uid = pla->la_uid;
2834         tla->la_gid = pla->la_gid;
2835         rc = dt_declare_attr_set(env, child, tla, handle);
2836         if (rc != 0)
2837                 GOTO(stop, rc);
2838
2839         rc = dt_trans_start(env, dev, handle);
2840         if (rc != 0)
2841                 GOTO(stop, rc);
2842
2843         dt_write_lock(env, parent, 0);
2844         if (unlikely(lu_object_is_dying(parent->do_lu.lo_header)))
2845                 GOTO(unlock2, rc = 1);
2846
2847         rc = dt_xattr_set(env, child, buf, XATTR_NAME_FID, 0, handle,
2848                           BYPASS_CAPA);
2849         if (rc != 0)
2850                 GOTO(unlock2, rc);
2851
2852         /* Get the latest parent's owner. */
2853         rc = dt_attr_get(env, parent, tla, BYPASS_CAPA);
2854         if (rc != 0)
2855                 GOTO(unlock2, rc);
2856
2857         tla->la_valid = LA_UID | LA_GID;
2858         rc = dt_attr_set(env, child, tla, handle, BYPASS_CAPA);
2859
2860         GOTO(unlock2, rc);
2861
2862 unlock2:
2863         dt_write_unlock(env, parent);
2864
2865 stop:
2866         rc = lfsck_layout_trans_stop(env, dev, handle, rc);
2867
2868 unlock1:
2869         lfsck_layout_unlock(&lh);
2870
2871         return rc;
2872 }
2873
2874 /* If there are more than one MDT-objects claim as the OST-object's parent,
2875  * and the OST-object only recognizes one of them, then we need to generate
2876  * new OST-object(s) with new fid(s) for the non-recognized MDT-object(s). */
2877 static int lfsck_layout_repair_multiple_references(const struct lu_env *env,
2878                                                    struct lfsck_component *com,
2879                                                    struct lfsck_layout_req *llr,
2880                                                    struct lu_attr *la,
2881                                                    struct lu_buf *buf)
2882 {
2883         struct lfsck_thread_info        *info   = lfsck_env_info(env);
2884         struct dt_allocation_hint       *hint   = &info->lti_hint;
2885         struct dt_object_format         *dof    = &info->lti_dof;
2886         struct dt_device                *pdev   = com->lc_lfsck->li_next;
2887         struct ost_id                   *oi     = &info->lti_oi;
2888         struct dt_object                *parent = llr->llr_parent->llo_obj;
2889         struct dt_device                *cdev   = lfsck_obj2dt_dev(llr->llr_child);
2890         struct dt_object                *child  = NULL;
2891         struct lu_device                *d      = &cdev->dd_lu_dev;
2892         struct lu_object                *o      = NULL;
2893         struct thandle                  *handle;
2894         struct lov_mds_md_v1            *lmm;
2895         struct lov_ost_data_v1          *objs;
2896         struct lustre_handle             lh     = { 0 };
2897         __u32                            magic;
2898         int                              rc;
2899         ENTRY;
2900
2901         CDEBUG(D_LFSCK, "Repair multiple references for: parent "DFID
2902                ", OST-index %u, stripe-index %u, owner %u:%u\n",
2903                PFID(lfsck_dto2fid(parent)), llr->llr_ost_idx,
2904                llr->llr_lov_idx, la->la_uid, la->la_gid);
2905
2906         rc = lfsck_layout_lock(env, com, parent, &lh,
2907                                MDS_INODELOCK_LAYOUT | MDS_INODELOCK_XATTR);
2908         if (rc != 0)
2909                 RETURN(rc);
2910
2911         handle = dt_trans_create(env, pdev);
2912         if (IS_ERR(handle))
2913                 GOTO(unlock1, rc = PTR_ERR(handle));
2914
2915         o = lu_object_anon(env, d, NULL);
2916         if (IS_ERR(o))
2917                 GOTO(stop, rc = PTR_ERR(o));
2918
2919         child = container_of(o, struct dt_object, do_lu);
2920         o = lu_object_locate(o->lo_header, d->ld_type);
2921         if (unlikely(o == NULL))
2922                 GOTO(stop, rc = -EINVAL);
2923
2924         child = container_of(o, struct dt_object, do_lu);
2925         la->la_valid = LA_UID | LA_GID;
2926         hint->dah_parent = NULL;
2927         hint->dah_mode = 0;
2928         dof->dof_type = DFT_REGULAR;
2929         rc = dt_declare_create(env, child, la, NULL, NULL, handle);
2930         if (rc != 0)
2931                 GOTO(stop, rc);
2932
2933         rc = dt_declare_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2934                                   LU_XATTR_REPLACE, handle);
2935         if (rc != 0)
2936                 GOTO(stop, rc);
2937
2938         rc = dt_trans_start(env, pdev, handle);
2939         if (rc != 0)
2940                 GOTO(stop, rc);
2941
2942         dt_write_lock(env, parent, 0);
2943         if (unlikely(lu_object_is_dying(parent->do_lu.lo_header)))
2944                 GOTO(unlock2, rc = 0);
2945
2946         rc = dt_xattr_get(env, parent, buf, XATTR_NAME_LOV, BYPASS_CAPA);
2947         if (unlikely(rc == 0 || rc == -ENODATA || rc == -ERANGE))
2948                 GOTO(unlock2, rc = 0);
2949
2950         lmm = buf->lb_buf;
2951         rc = lfsck_layout_verify_header(lmm);
2952         if (rc != 0)
2953                 GOTO(unlock2, rc);
2954
2955         /* Someone change layout during the LFSCK, no need to repair then. */
2956         if (le16_to_cpu(lmm->lmm_layout_gen) != llr->llr_parent->llo_gen)
2957                 GOTO(unlock2, rc = 0);
2958
2959         rc = dt_create(env, child, la, hint, dof, handle);
2960         if (rc != 0)
2961                 GOTO(unlock2, rc);
2962
2963         /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
2964          * been verified in lfsck_layout_verify_header() already. If some
2965          * new magic introduced in the future, then layout LFSCK needs to
2966          * be updated also. */
2967         magic = le32_to_cpu(lmm->lmm_magic);
2968         if (magic == LOV_MAGIC_V1) {
2969                 objs = &(lmm->lmm_objects[0]);
2970         } else {
2971                 LASSERT(magic == LOV_MAGIC_V3);
2972                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
2973         }
2974
2975         lmm->lmm_layout_gen = cpu_to_le16(llr->llr_parent->llo_gen + 1);
2976         fid_to_ostid(lu_object_fid(&child->do_lu), oi);
2977         ostid_cpu_to_le(oi, &objs[llr->llr_lov_idx].l_ost_oi);
2978         objs[llr->llr_lov_idx].l_ost_gen = cpu_to_le32(0);
2979         objs[llr->llr_lov_idx].l_ost_idx = cpu_to_le32(llr->llr_ost_idx);
2980         rc = dt_xattr_set(env, parent, buf, XATTR_NAME_LOV,
2981                           LU_XATTR_REPLACE, handle, BYPASS_CAPA);
2982
2983         GOTO(unlock2, rc = (rc == 0 ? 1 : rc));
2984
2985 unlock2:
2986         dt_write_unlock(env, parent);
2987
2988 stop:
2989         if (child != NULL)
2990                 lu_object_put(env, &child->do_lu);
2991
2992         dt_trans_stop(env, pdev, handle);
2993
2994 unlock1:
2995         lfsck_layout_unlock(&lh);
2996
2997         return rc;
2998 }
2999
3000 /* If the MDT-object and the OST-object have different owner information,
3001  * then trust the MDT-object, because the normal chown/chgrp handle order
3002  * is from MDT to OST, and it is possible that some chown/chgrp operation
3003  * is partly done. */
3004 static int lfsck_layout_repair_owner(const struct lu_env *env,
3005                                      struct lfsck_component *com,
3006                                      struct lfsck_layout_req *llr,
3007                                      struct lu_attr *pla)
3008 {
3009         struct lfsck_thread_info        *info   = lfsck_env_info(env);
3010         struct lu_attr                  *tla    = &info->lti_la3;
3011         struct dt_object                *parent = llr->llr_parent->llo_obj;
3012         struct dt_object                *child  = llr->llr_child;
3013         struct dt_device                *dev    = lfsck_obj2dt_dev(child);
3014         struct thandle                  *handle;
3015         int                              rc;
3016         ENTRY;
3017
3018         CDEBUG(D_LFSCK, "Repair inconsistent file owner for: parent "DFID
3019                ", child "DFID", OST-index %u, stripe-index %u, owner %u:%u\n",
3020                PFID(lfsck_dto2fid(parent)), PFID(lfsck_dto2fid(child)),
3021                llr->llr_ost_idx, llr->llr_lov_idx, pla->la_uid, pla->la_gid);
3022
3023         handle = dt_trans_create(env, dev);
3024         if (IS_ERR(handle))
3025                 RETURN(PTR_ERR(handle));
3026
3027         tla->la_uid = pla->la_uid;
3028         tla->la_gid = pla->la_gid;
3029         tla->la_valid = LA_UID | LA_GID;
3030         rc = dt_declare_attr_set(env, child, tla, handle);
3031         if (rc != 0)
3032                 GOTO(stop, rc);
3033
3034         rc = dt_trans_start(env, dev, handle);
3035         if (rc != 0)
3036                 GOTO(stop, rc);
3037
3038         /* Use the dt_object lock to serialize with destroy and attr_set. */
3039         dt_read_lock(env, parent, 0);
3040         if (unlikely(lu_object_is_dying(parent->do_lu.lo_header)))
3041                 GOTO(unlock, rc = 1);
3042
3043         /* Get the latest parent's owner. */
3044         rc = dt_attr_get(env, parent, tla, BYPASS_CAPA);
3045         if (rc != 0) {
3046                 CWARN("%s: fail to get the latest parent's ("DFID") owner, "
3047                       "not sure whether some others chown/chgrp during the "
3048                       "LFSCK: rc = %d\n", lfsck_lfsck2name(com->lc_lfsck),
3049                       PFID(lfsck_dto2fid(parent)), rc);
3050
3051                 GOTO(unlock, rc);
3052         }
3053
3054         /* Some others chown/chgrp during the LFSCK, needs to do nothing. */
3055         if (unlikely(tla->la_uid != pla->la_uid ||
3056                      tla->la_gid != pla->la_gid))
3057                 GOTO(unlock, rc = 1);
3058
3059         tla->la_valid = LA_UID | LA_GID;
3060         rc = dt_attr_set(env, child, tla, handle, BYPASS_CAPA);
3061
3062         GOTO(unlock, rc);
3063
3064 unlock:
3065         dt_read_unlock(env, parent);
3066
3067 stop:
3068         rc = lfsck_layout_trans_stop(env, dev, handle, rc);
3069
3070         return rc;
3071 }
3072
3073 /* Check whether the OST-object correctly back points to the
3074  * MDT-object (@parent) via the XATTR_NAME_FID xattr (@pfid). */
3075 static int lfsck_layout_check_parent(const struct lu_env *env,
3076                                      struct lfsck_component *com,
3077                                      struct dt_object *parent,
3078                                      const struct lu_fid *pfid,
3079                                      const struct lu_fid *cfid,
3080                                      const struct lu_attr *pla,
3081                                      const struct lu_attr *cla,
3082                                      struct lfsck_layout_req *llr,
3083                                      struct lu_buf *lov_ea, __u32 idx)
3084 {
3085         struct lfsck_thread_info        *info   = lfsck_env_info(env);
3086         struct lu_buf                   *buf    = &info->lti_big_buf;
3087         struct dt_object                *tobj;
3088         struct lov_mds_md_v1            *lmm;
3089         struct lov_ost_data_v1          *objs;
3090         int                              rc;
3091         int                              i;
3092         __u32                            magic;
3093         __u16                            count;
3094         ENTRY;
3095
3096         if (fid_is_zero(pfid)) {
3097                 /* client never wrote. */
3098                 if (cla->la_size == 0 && cla->la_blocks == 0) {
3099                         if (unlikely(cla->la_uid != pla->la_uid ||
3100                                      cla->la_gid != pla->la_gid))
3101                                 RETURN (LLIT_INCONSISTENT_OWNER);
3102
3103                         RETURN(0);
3104                 }
3105
3106                 RETURN(LLIT_UNMATCHED_PAIR);
3107         }
3108
3109         if (unlikely(!fid_is_sane(pfid)))
3110                 RETURN(LLIT_UNMATCHED_PAIR);
3111
3112         if (lu_fid_eq(pfid, lu_object_fid(&parent->do_lu))) {
3113                 if (llr->llr_lov_idx == idx)
3114                         RETURN(0);
3115
3116                 RETURN(LLIT_UNMATCHED_PAIR);
3117         }
3118
3119         tobj = lfsck_object_find(env, com->lc_lfsck, pfid);
3120         if (tobj == NULL)
3121                 RETURN(LLIT_UNMATCHED_PAIR);
3122
3123         if (IS_ERR(tobj))
3124                 RETURN(PTR_ERR(tobj));
3125
3126         if (!dt_object_exists(tobj))
3127                 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3128
3129         /* Load the tobj's layout EA, in spite of it is a local MDT-object or
3130          * remote one on another MDT. Then check whether the given OST-object
3131          * is in such layout. If yes, it is multiple referenced, otherwise it
3132          * is unmatched referenced case. */
3133         rc = lfsck_layout_get_lovea(env, tobj, buf, NULL);
3134         if (rc == 0)
3135                 GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3136
3137         if (rc < 0)
3138                 GOTO(out, rc);
3139
3140         lmm = buf->lb_buf;
3141         rc = lfsck_layout_verify_header(lmm);
3142         if (rc != 0)
3143                 GOTO(out, rc);
3144
3145         /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
3146          * been verified in lfsck_layout_verify_header() already. If some
3147          * new magic introduced in the future, then layout LFSCK needs to
3148          * be updated also. */
3149         magic = le32_to_cpu(lmm->lmm_magic);
3150         if (magic == LOV_MAGIC_V1) {
3151                 objs = &(lmm->lmm_objects[0]);
3152         } else {
3153                 LASSERT(magic == LOV_MAGIC_V3);
3154                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
3155         }
3156
3157         count = le16_to_cpu(lmm->lmm_stripe_count);
3158         for (i = 0; i < count; i++, objs++) {
3159                 struct lu_fid           *tfid   = &info->lti_fid2;
3160                 struct ost_id           *oi     = &info->lti_oi;
3161
3162                 if (is_dummy_lov_ost_data(objs))
3163                         continue;
3164
3165                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
3166                 ostid_to_fid(tfid, oi, le32_to_cpu(objs->l_ost_idx));
3167                 if (lu_fid_eq(cfid, tfid)) {
3168                         *lov_ea = *buf;
3169
3170                         GOTO(out, rc = LLIT_MULTIPLE_REFERENCED);
3171                 }
3172         }
3173
3174         GOTO(out, rc = LLIT_UNMATCHED_PAIR);
3175
3176 out:
3177         lfsck_object_put(env, tobj);
3178
3179         return rc;
3180 }
3181
3182 static int lfsck_layout_assistant_handle_one(const struct lu_env *env,
3183                                              struct lfsck_component *com,
3184                                              struct lfsck_layout_req *llr)
3185 {
3186         struct lfsck_layout                  *lo     = com->lc_file_ram;
3187         struct lfsck_thread_info             *info   = lfsck_env_info(env);
3188         struct filter_fid_old                *pea    = &info->lti_old_pfid;
3189         struct lu_fid                        *pfid   = &info->lti_fid;
3190         struct lu_buf                        *buf    = NULL;
3191         struct dt_object                     *parent = llr->llr_parent->llo_obj;
3192         struct dt_object                     *child  = llr->llr_child;
3193         struct lu_attr                       *pla    = &info->lti_la;
3194         struct lu_attr                       *cla    = &info->lti_la2;
3195         struct lfsck_instance                *lfsck  = com->lc_lfsck;
3196         struct lfsck_bookmark                *bk     = &lfsck->li_bookmark_ram;
3197         enum lfsck_layout_inconsistency_type  type   = LLIT_NONE;
3198         __u32                                 idx    = 0;
3199         int                                   rc;
3200         ENTRY;
3201
3202         rc = dt_attr_get(env, parent, pla, BYPASS_CAPA);
3203         if (rc != 0) {
3204                 if (lu_object_is_dying(parent->do_lu.lo_header))
3205                         RETURN(0);
3206
3207                 GOTO(out, rc);
3208         }
3209
3210         rc = dt_attr_get(env, child, cla, BYPASS_CAPA);
3211         if (rc == -ENOENT) {
3212                 if (lu_object_is_dying(parent->do_lu.lo_header))
3213                         RETURN(0);
3214
3215                 type = LLIT_DANGLING;
3216                 goto repair;
3217         }
3218
3219         if (rc != 0)
3220                 GOTO(out, rc);
3221
3222         buf = lfsck_buf_get(env, pea, sizeof(struct filter_fid_old));
3223         rc= dt_xattr_get(env, child, buf, XATTR_NAME_FID, BYPASS_CAPA);
3224         if (unlikely(rc >= 0 && rc != sizeof(struct filter_fid_old) &&
3225                      rc != sizeof(struct filter_fid))) {
3226                 type = LLIT_UNMATCHED_PAIR;
3227                 goto repair;
3228         }
3229
3230         if (rc < 0 && rc != -ENODATA)
3231                 GOTO(out, rc);
3232
3233         if (rc == -ENODATA) {
3234                 fid_zero(pfid);
3235         } else {
3236                 fid_le_to_cpu(pfid, &pea->ff_parent);
3237                 /* OST-object does not save parent FID::f_ver, instead,
3238                  * the OST-object index in the parent MDT-object layout
3239                  * EA reuses the pfid->f_ver. */
3240                 idx = pfid->f_ver;
3241                 pfid->f_ver = 0;
3242         }
3243
3244         rc = lfsck_layout_check_parent(env, com, parent, pfid,
3245                                        lu_object_fid(&child->do_lu),
3246                                        pla, cla, llr, buf, idx);
3247         if (rc > 0) {
3248                 type = rc;
3249                 goto repair;
3250         }
3251
3252         if (rc < 0)
3253                 GOTO(out, rc);
3254
3255         if (unlikely(cla->la_uid != pla->la_uid ||
3256                      cla->la_gid != pla->la_gid)) {
3257                 type = LLIT_INCONSISTENT_OWNER;
3258                 goto repair;
3259         }
3260
3261 repair:
3262         if (bk->lb_param & LPF_DRYRUN) {
3263                 if (type != LLIT_NONE)
3264                         GOTO(out, rc = 1);
3265                 else
3266                         GOTO(out, rc = 0);
3267         }
3268
3269         switch (type) {
3270         case LLIT_DANGLING:
3271                 memset(cla, 0, sizeof(*cla));
3272                 cla->la_uid = pla->la_uid;
3273                 cla->la_gid = pla->la_gid;
3274                 cla->la_mode = S_IFREG | 0666;
3275                 cla->la_valid = LA_TYPE | LA_MODE | LA_UID | LA_GID |
3276                                 LA_ATIME | LA_MTIME | LA_CTIME;
3277                 rc = lfsck_layout_recreate_ostobj(env, com, llr, cla);
3278                 break;
3279         case LLIT_UNMATCHED_PAIR:
3280                 rc = lfsck_layout_repair_unmatched_pair(env, com, llr, pla);
3281                 break;
3282         case LLIT_MULTIPLE_REFERENCED:
3283                 rc = lfsck_layout_repair_multiple_references(env, com, llr,
3284                                                              pla, buf);
3285                 break;
3286         case LLIT_INCONSISTENT_OWNER:
3287                 rc = lfsck_layout_repair_owner(env, com, llr, pla);
3288                 break;
3289         default:
3290                 rc = 0;
3291                 break;
3292         }
3293
3294         GOTO(out, rc);
3295
3296 out:
3297         down_write(&com->lc_sem);
3298         if (rc < 0) {
3299                 /* If cannot touch the target server,
3300                  * mark the LFSCK as INCOMPLETE. */
3301                 if (rc == -ENOTCONN || rc == -ESHUTDOWN || rc == -ETIMEDOUT ||
3302                     rc == -EHOSTDOWN || rc == -EHOSTUNREACH) {
3303                         CERROR("%s: Fail to talk with OST %x: rc = %d.\n",
3304                                lfsck_lfsck2name(lfsck), llr->llr_ost_idx, rc);
3305                         lo->ll_flags |= LF_INCOMPLETE;
3306                         lo->ll_objs_skipped++;
3307                         rc = 0;
3308                 } else {
3309                         lo->ll_objs_failed_phase1++;
3310                 }
3311         } else if (rc > 0) {
3312                 LASSERTF(type > LLIT_NONE && type <= LLIT_MAX,
3313                          "unknown type = %d\n", type);
3314
3315                 lo->ll_objs_repaired[type - 1]++;
3316         }
3317         up_write(&com->lc_sem);
3318
3319         return rc;
3320 }
3321
3322 static int lfsck_layout_assistant(void *args)
3323 {
3324         struct lfsck_thread_args        *lta     = args;
3325         struct lu_env                   *env     = &lta->lta_env;
3326         struct lfsck_component          *com     = lta->lta_com;
3327         struct lfsck_instance           *lfsck   = lta->lta_lfsck;
3328         struct lfsck_bookmark           *bk      = &lfsck->li_bookmark_ram;
3329         struct lfsck_position           *pos     = &com->lc_pos_start;
3330         struct lfsck_thread_info        *info    = lfsck_env_info(env);
3331         struct lfsck_request            *lr      = &info->lti_lr;
3332         struct lfsck_layout_master_data *llmd    = com->lc_data;
3333         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
3334         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
3335         struct lfsck_layout_req         *llr;
3336         struct l_wait_info               lwi     = { 0 };
3337         int                              rc      = 0;
3338         int                              rc1     = 0;
3339         ENTRY;
3340
3341         memset(lr, 0, sizeof(*lr));
3342         lr->lr_event = LE_START;
3343         lr->lr_valid = LSV_SPEED_LIMIT | LSV_ERROR_HANDLE | LSV_DRYRUN |
3344                        LSV_ASYNC_WINDOWS;
3345         lr->lr_speed = bk->lb_speed_limit;
3346         lr->lr_version = bk->lb_version;
3347         lr->lr_param = bk->lb_param;
3348         lr->lr_async_windows = bk->lb_async_windows;
3349         lr->lr_flags = LEF_TO_OST;
3350         if (pos->lp_oit_cookie <= 1)
3351                 lr->lr_param |= LPF_RESET;
3352
3353         rc = lfsck_layout_master_notify_others(env, com, lr);
3354         if (rc != 0) {
3355                 CERROR("%s: fail to notify others for layout start: rc = %d\n",
3356                        lfsck_lfsck2name(lfsck), rc);
3357                 GOTO(fini, rc);
3358         }
3359
3360         spin_lock(&llmd->llmd_lock);
3361         thread_set_flags(athread, SVC_RUNNING);
3362         spin_unlock(&llmd->llmd_lock);
3363         wake_up_all(&mthread->t_ctl_waitq);
3364
3365         while (1) {
3366                 while (!list_empty(&llmd->llmd_req_list)) {
3367                         bool wakeup = false;
3368
3369                         if (unlikely(llmd->llmd_exit))
3370                                 GOTO(cleanup1, rc = llmd->llmd_post_result);
3371
3372                         llr = list_entry(llmd->llmd_req_list.next,
3373                                          struct lfsck_layout_req,
3374                                          llr_list);
3375                         /* Only the lfsck_layout_assistant thread itself can
3376                          * remove the "llr" from the head of the list, LFSCK
3377                          * engine thread only inserts other new "lld" at the
3378                          * end of the list. So it is safe to handle current
3379                          * "llr" without the spin_lock. */
3380                         rc = lfsck_layout_assistant_handle_one(env, com, llr);
3381                         spin_lock(&llmd->llmd_lock);
3382                         list_del_init(&llr->llr_list);
3383                         llmd->llmd_prefetched--;
3384                         /* Wake up the main engine thread only when the list
3385                          * is empty or half of the prefetched items have been
3386                          * handled to avoid too frequent thread schedule. */
3387                         if (llmd->llmd_prefetched == 0 ||
3388                             (bk->lb_async_windows != 0 &&
3389                              (bk->lb_async_windows >> 1) ==
3390                              llmd->llmd_prefetched))
3391                                 wakeup = true;
3392                         spin_unlock(&llmd->llmd_lock);
3393                         if (wakeup)
3394                                 wake_up_all(&mthread->t_ctl_waitq);
3395
3396                         lfsck_layout_req_fini(env, llr);
3397                         if (rc < 0 && bk->lb_param & LPF_FAILOUT)
3398                                 GOTO(cleanup1, rc);
3399                 }
3400
3401                 l_wait_event(athread->t_ctl_waitq,
3402                              !lfsck_layout_req_empty(llmd) ||
3403                              llmd->llmd_exit ||
3404                              llmd->llmd_to_post ||
3405                              llmd->llmd_to_double_scan,
3406                              &lwi);
3407
3408                 if (unlikely(llmd->llmd_exit))
3409                         GOTO(cleanup1, rc = llmd->llmd_post_result);
3410
3411                 if (!list_empty(&llmd->llmd_req_list))
3412                         continue;
3413
3414                 if (llmd->llmd_to_post) {
3415                         llmd->llmd_to_post = 0;
3416                         LASSERT(llmd->llmd_post_result > 0);
3417
3418                         memset(lr, 0, sizeof(*lr));
3419                         lr->lr_event = LE_PHASE1_DONE;
3420                         lr->lr_status = llmd->llmd_post_result;
3421                         rc = lfsck_layout_master_notify_others(env, com, lr);
3422                         if (rc != 0)
3423                                 CERROR("%s: failed to notify others "
3424                                        "for layout post: rc = %d\n",
3425                                        lfsck_lfsck2name(lfsck), rc);
3426
3427                         /* Wakeup the master engine to go ahead. */
3428                         wake_up_all(&mthread->t_ctl_waitq);
3429                 }
3430
3431                 if (llmd->llmd_to_double_scan) {
3432                         llmd->llmd_to_double_scan = 0;
3433                         atomic_inc(&lfsck->li_double_scan_count);
3434                         llmd->llmd_in_double_scan = 1;
3435                         wake_up_all(&mthread->t_ctl_waitq);
3436
3437                         com->lc_new_checked = 0;
3438                         com->lc_new_scanned = 0;
3439                         com->lc_time_last_checkpoint = cfs_time_current();
3440                         com->lc_time_next_checkpoint =
3441                                 com->lc_time_last_checkpoint +
3442                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
3443
3444                         while (llmd->llmd_in_double_scan) {
3445                                 struct lfsck_tgt_descs  *ltds =
3446                                                         &lfsck->li_ost_descs;
3447                                 struct lfsck_tgt_desc   *ltd;
3448
3449                                 rc = lfsck_layout_master_query_others(env, com);
3450                                 if (lfsck_layout_master_to_orphan(llmd))
3451                                         goto orphan;
3452
3453                                 if (rc < 0)
3454                                         GOTO(cleanup2, rc);
3455
3456                                 /* Pull LFSCK status on related targets once
3457                                  * per 30 seconds if we are not notified. */
3458                                 lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(30),
3459                                                            cfs_time_seconds(1),
3460                                                            NULL, NULL);
3461                                 rc = l_wait_event(athread->t_ctl_waitq,
3462                                         lfsck_layout_master_to_orphan(llmd) ||
3463                                         llmd->llmd_exit ||
3464                                         !thread_is_running(mthread),
3465                                         &lwi);
3466
3467                                 if (unlikely(llmd->llmd_exit ||
3468                                              !thread_is_running(mthread)))
3469                                         GOTO(cleanup2, rc = 0);
3470
3471                                 if (rc == -ETIMEDOUT)
3472                                         continue;
3473
3474                                 if (rc < 0)
3475                                         GOTO(cleanup2, rc);
3476
3477 orphan:
3478                                 spin_lock(&ltds->ltd_lock);
3479                                 while (!list_empty(
3480                                                 &llmd->llmd_ost_phase2_list)) {
3481                                         ltd = list_entry(
3482                                               llmd->llmd_ost_phase2_list.next,
3483                                               struct lfsck_tgt_desc,
3484                                               ltd_layout_phase_list);
3485                                         list_del_init(
3486                                                 &ltd->ltd_layout_phase_list);
3487                                         spin_unlock(&ltds->ltd_lock);
3488
3489                                         if (bk->lb_param & LPF_ALL_TGT) {
3490                                                 rc = lfsck_layout_scan_orphan(
3491                                                                 env, com, ltd);
3492                                                 if (rc != 0 &&
3493                                                     bk->lb_param & LPF_FAILOUT)
3494                                                         GOTO(cleanup2, rc);
3495                                         }
3496
3497                                         if (unlikely(llmd->llmd_exit ||
3498                                                 !thread_is_running(mthread)))
3499                                                 GOTO(cleanup2, rc = 0);
3500
3501                                         spin_lock(&ltds->ltd_lock);
3502                                 }
3503
3504                                 if (list_empty(&llmd->llmd_ost_phase1_list)) {
3505                                         spin_unlock(&ltds->ltd_lock);
3506                                         GOTO(cleanup2, rc = 1);
3507                                 }
3508                                 spin_unlock(&ltds->ltd_lock);
3509                         }
3510                 }
3511         }
3512
3513 cleanup1:
3514         /* Cleanup the unfinished requests. */
3515         spin_lock(&llmd->llmd_lock);
3516         if (rc < 0)
3517                 llmd->llmd_assistant_status = rc;
3518
3519         while (!list_empty(&llmd->llmd_req_list)) {
3520                 llr = list_entry(llmd->llmd_req_list.next,
3521                                  struct lfsck_layout_req,
3522                                  llr_list);
3523                 list_del_init(&llr->llr_list);
3524                 llmd->llmd_prefetched--;
3525                 spin_unlock(&llmd->llmd_lock);
3526                 lfsck_layout_req_fini(env, llr);
3527                 spin_lock(&llmd->llmd_lock);
3528         }
3529         spin_unlock(&llmd->llmd_lock);
3530
3531         LASSERTF(llmd->llmd_prefetched == 0, "unmatched prefeteched objs %d\n",
3532                  llmd->llmd_prefetched);
3533
3534 cleanup2:
3535         memset(lr, 0, sizeof(*lr));
3536         if (rc > 0) {
3537                 lr->lr_event = LE_PHASE2_DONE;
3538                 lr->lr_status = rc;
3539         } else if (rc == 0) {
3540                 if (lfsck->li_flags & LPF_ALL_TGT) {
3541                         lr->lr_event = LE_STOP;
3542                         lr->lr_status = LS_STOPPED;
3543                 } else {
3544                         lr->lr_event = LE_PEER_EXIT;
3545                         switch (lfsck->li_status) {
3546                         case LS_PAUSED:
3547                         case LS_CO_PAUSED:
3548                                 lr->lr_status = LS_CO_PAUSED;
3549                                 break;
3550                         case LS_STOPPED:
3551                         case LS_CO_STOPPED:
3552                                 lr->lr_status = LS_CO_STOPPED;
3553                                 break;
3554                         default:
3555                                 CERROR("%s: unknown status: rc = %d\n",
3556                                        lfsck_lfsck2name(lfsck),
3557                                        lfsck->li_status);
3558                                 lr->lr_status = LS_CO_FAILED;
3559                                 break;
3560                         }
3561                 }
3562         } else {
3563                 if (lfsck->li_flags & LPF_ALL_TGT) {
3564                         lr->lr_event = LE_STOP;
3565                         lr->lr_status = LS_FAILED;
3566                 } else {
3567                         lr->lr_event = LE_PEER_EXIT;
3568                         lr->lr_status = LS_CO_FAILED;
3569                 }
3570         }
3571
3572         rc1 = lfsck_layout_master_notify_others(env, com, lr);
3573         if (rc1 != 0) {
3574                 CERROR("%s: failed to notify others for layout quit: rc = %d\n",
3575                        lfsck_lfsck2name(lfsck), rc1);
3576                 rc = rc1;
3577         }
3578
3579         /* Under force exit case, some requests may be just freed without
3580          * verification, those objects should be re-handled when next run.
3581          * So not update the on-disk tracing file under such case. */
3582         if (!llmd->llmd_exit)
3583                 rc1 = lfsck_layout_double_scan_result(env, com, rc);
3584
3585 fini:
3586         if (llmd->llmd_in_double_scan)
3587                 atomic_dec(&lfsck->li_double_scan_count);
3588
3589         spin_lock(&llmd->llmd_lock);
3590         llmd->llmd_assistant_status = (rc1 != 0 ? rc1 : rc);
3591         thread_set_flags(athread, SVC_STOPPED);
3592         wake_up_all(&mthread->t_ctl_waitq);
3593         spin_unlock(&llmd->llmd_lock);
3594         lfsck_thread_args_fini(lta);
3595
3596         return rc;
3597 }
3598
3599 static int
3600 lfsck_layout_slave_async_interpret(const struct lu_env *env,
3601                                    struct ptlrpc_request *req,
3602                                    void *args, int rc)
3603 {
3604         struct lfsck_layout_slave_async_args *llsaa = args;
3605         struct obd_export                    *exp   = llsaa->llsaa_exp;
3606         struct lfsck_component               *com   = llsaa->llsaa_com;
3607         struct lfsck_layout_slave_target     *llst  = llsaa->llsaa_llst;
3608         struct lfsck_layout_slave_data       *llsd  = com->lc_data;
3609         bool                                  done  = false;
3610
3611         if (rc != 0) {
3612                 /* It is quite probably caused by target crash,
3613                  * to make the LFSCK can go ahead, assume that
3614                  * the target finished the LFSCK prcoessing. */
3615                 done = true;
3616         } else {
3617                 struct lfsck_reply *lr;
3618
3619                 lr = req_capsule_server_get(&req->rq_pill, &RMF_LFSCK_REPLY);
3620                 if (lr->lr_status != LS_SCANNING_PHASE1 &&
3621                     lr->lr_status != LS_SCANNING_PHASE2)
3622                         done = true;
3623         }
3624         if (done)
3625                 lfsck_layout_llst_del(llsd, llst);
3626         lfsck_layout_llst_put(llst);
3627         lfsck_component_put(env, com);
3628         class_export_put(exp);
3629
3630         return 0;
3631 }
3632
3633 static int lfsck_layout_async_query(const struct lu_env *env,
3634                                     struct lfsck_component *com,
3635                                     struct obd_export *exp,
3636                                     struct lfsck_layout_slave_target *llst,
3637                                     struct lfsck_request *lr,
3638                                     struct ptlrpc_request_set *set)
3639 {
3640         struct lfsck_layout_slave_async_args *llsaa;
3641         struct ptlrpc_request                *req;
3642         struct lfsck_request                 *tmp;
3643         int                                   rc;
3644         ENTRY;
3645
3646         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_QUERY);
3647         if (req == NULL)
3648                 RETURN(-ENOMEM);
3649
3650         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_QUERY);
3651         if (rc != 0) {
3652                 ptlrpc_request_free(req);
3653                 RETURN(rc);
3654         }
3655
3656         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3657         *tmp = *lr;
3658         ptlrpc_request_set_replen(req);
3659
3660         llsaa = ptlrpc_req_async_args(req);
3661         llsaa->llsaa_exp = exp;
3662         llsaa->llsaa_com = lfsck_component_get(com);
3663         llsaa->llsaa_llst = llst;
3664         req->rq_interpret_reply = lfsck_layout_slave_async_interpret;
3665         ptlrpc_set_add_req(set, req);
3666
3667         RETURN(0);
3668 }
3669
3670 static int lfsck_layout_async_notify(const struct lu_env *env,
3671                                      struct obd_export *exp,
3672                                      struct lfsck_request *lr,
3673                                      struct ptlrpc_request_set *set)
3674 {
3675         struct ptlrpc_request   *req;
3676         struct lfsck_request    *tmp;
3677         int                      rc;
3678         ENTRY;
3679
3680         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3681         if (req == NULL)
3682                 RETURN(-ENOMEM);
3683
3684         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3685         if (rc != 0) {
3686                 ptlrpc_request_free(req);
3687                 RETURN(rc);
3688         }
3689
3690         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3691         *tmp = *lr;
3692         ptlrpc_request_set_replen(req);
3693         ptlrpc_set_add_req(set, req);
3694
3695         RETURN(0);
3696 }
3697
3698 static int
3699 lfsck_layout_slave_query_master(const struct lu_env *env,
3700                                 struct lfsck_component *com)
3701 {
3702         struct lfsck_request             *lr    = &lfsck_env_info(env)->lti_lr;
3703         struct lfsck_instance            *lfsck = com->lc_lfsck;
3704         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
3705         struct lfsck_layout_slave_target *llst;
3706         struct obd_export                *exp;
3707         struct ptlrpc_request_set        *set;
3708         int                               rc    = 0;
3709         int                               rc1   = 0;
3710         ENTRY;
3711
3712         set = ptlrpc_prep_set();
3713         if (set == NULL)
3714                 RETURN(-ENOMEM);
3715
3716         memset(lr, 0, sizeof(*lr));
3717         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
3718         lr->lr_event = LE_QUERY;
3719         lr->lr_active = LT_LAYOUT;
3720
3721         llsd->llsd_touch_gen++;
3722         spin_lock(&llsd->llsd_lock);
3723         while (!list_empty(&llsd->llsd_master_list)) {
3724                 llst = list_entry(llsd->llsd_master_list.next,
3725                                   struct lfsck_layout_slave_target,
3726                                   llst_list);
3727                 if (llst->llst_gen == llsd->llsd_touch_gen)
3728                         break;
3729
3730                 llst->llst_gen = llsd->llsd_touch_gen;
3731                 list_del(&llst->llst_list);
3732                 list_add_tail(&llst->llst_list,
3733                               &llsd->llsd_master_list);
3734                 atomic_inc(&llst->llst_ref);
3735                 spin_unlock(&llsd->llsd_lock);
3736
3737                 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3738                                                llst->llst_index);
3739                 if (exp == NULL) {
3740                         lfsck_layout_llst_del(llsd, llst);
3741                         lfsck_layout_llst_put(llst);
3742                         spin_lock(&llsd->llsd_lock);
3743                         continue;
3744                 }
3745
3746                 rc = lfsck_layout_async_query(env, com, exp, llst, lr, set);
3747                 if (rc != 0) {
3748                         CERROR("%s: slave fail to query %s for layout: "
3749                                "rc = %d\n", lfsck_lfsck2name(lfsck),
3750                                exp->exp_obd->obd_name, rc);
3751                         rc1 = rc;
3752                         lfsck_layout_llst_put(llst);
3753                         class_export_put(exp);
3754                 }
3755                 spin_lock(&llsd->llsd_lock);
3756         }
3757         spin_unlock(&llsd->llsd_lock);
3758
3759         rc = ptlrpc_set_wait(set);
3760         ptlrpc_set_destroy(set);
3761
3762         RETURN(rc1 != 0 ? rc1 : rc);
3763 }
3764
3765 static void
3766 lfsck_layout_slave_notify_master(const struct lu_env *env,
3767                                  struct lfsck_component *com,
3768                                  enum lfsck_events event, int result)
3769 {
3770         struct lfsck_instance            *lfsck = com->lc_lfsck;
3771         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
3772         struct lfsck_request             *lr    = &lfsck_env_info(env)->lti_lr;
3773         struct lfsck_layout_slave_target *llst;
3774         struct obd_export                *exp;
3775         struct ptlrpc_request_set        *set;
3776         int                               rc;
3777         ENTRY;
3778
3779         set = ptlrpc_prep_set();
3780         if (set == NULL)
3781                 RETURN_EXIT;
3782
3783         memset(lr, 0, sizeof(*lr));
3784         lr->lr_event = event;
3785         lr->lr_flags = LEF_FROM_OST;
3786         lr->lr_status = result;
3787         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
3788         lr->lr_active = LT_LAYOUT;
3789         llsd->llsd_touch_gen++;
3790         spin_lock(&llsd->llsd_lock);
3791         while (!list_empty(&llsd->llsd_master_list)) {
3792                 llst = list_entry(llsd->llsd_master_list.next,
3793                                   struct lfsck_layout_slave_target,
3794                                   llst_list);
3795                 if (llst->llst_gen == llsd->llsd_touch_gen)
3796                         break;
3797
3798                 llst->llst_gen = llsd->llsd_touch_gen;
3799                 list_del(&llst->llst_list);
3800                 list_add_tail(&llst->llst_list,
3801                               &llsd->llsd_master_list);
3802                 atomic_inc(&llst->llst_ref);
3803                 spin_unlock(&llsd->llsd_lock);
3804
3805                 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3806                                                llst->llst_index);
3807                 if (exp == NULL) {
3808                         lfsck_layout_llst_del(llsd, llst);
3809                         lfsck_layout_llst_put(llst);
3810                         spin_lock(&llsd->llsd_lock);
3811                         continue;
3812                 }
3813
3814                 rc = lfsck_layout_async_notify(env, exp, lr, set);
3815                 if (rc != 0)
3816                         CERROR("%s: slave fail to notify %s for layout: "
3817                                "rc = %d\n", lfsck_lfsck2name(lfsck),
3818                                exp->exp_obd->obd_name, rc);
3819                 lfsck_layout_llst_put(llst);
3820                 class_export_put(exp);
3821                 spin_lock(&llsd->llsd_lock);
3822         }
3823         spin_unlock(&llsd->llsd_lock);
3824
3825         ptlrpc_set_wait(set);
3826         ptlrpc_set_destroy(set);
3827
3828         RETURN_EXIT;
3829 }
3830
3831 /* layout APIs */
3832
3833 static int lfsck_layout_reset(const struct lu_env *env,
3834                               struct lfsck_component *com, bool init)
3835 {
3836         struct lfsck_layout     *lo    = com->lc_file_ram;
3837         int                      rc;
3838
3839         down_write(&com->lc_sem);
3840         if (init) {
3841                 memset(lo, 0, com->lc_file_size);
3842         } else {
3843                 __u32 count = lo->ll_success_count;
3844                 __u64 last_time = lo->ll_time_last_complete;
3845
3846                 memset(lo, 0, com->lc_file_size);
3847                 lo->ll_success_count = count;
3848                 lo->ll_time_last_complete = last_time;
3849         }
3850
3851         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
3852         lo->ll_status = LS_INIT;
3853
3854         rc = lfsck_layout_store(env, com);
3855         up_write(&com->lc_sem);
3856
3857         return rc;
3858 }
3859
3860 static void lfsck_layout_fail(const struct lu_env *env,
3861                               struct lfsck_component *com, bool new_checked)
3862 {
3863         struct lfsck_layout *lo = com->lc_file_ram;
3864
3865         down_write(&com->lc_sem);
3866         if (new_checked)
3867                 com->lc_new_checked++;
3868         lo->ll_objs_failed_phase1++;
3869         if (lo->ll_pos_first_inconsistent == 0) {
3870                 struct lfsck_instance *lfsck = com->lc_lfsck;
3871
3872                 lo->ll_pos_first_inconsistent =
3873                         lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
3874                                                         lfsck->li_di_oit);
3875         }
3876         up_write(&com->lc_sem);
3877 }
3878
3879 static int lfsck_layout_master_checkpoint(const struct lu_env *env,
3880                                           struct lfsck_component *com, bool init)
3881 {
3882         struct lfsck_instance           *lfsck   = com->lc_lfsck;
3883         struct lfsck_layout             *lo      = com->lc_file_ram;
3884         struct lfsck_layout_master_data *llmd    = com->lc_data;
3885         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
3886         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
3887         struct l_wait_info               lwi     = { 0 };
3888         int                              rc;
3889
3890         if (com->lc_new_checked == 0 && !init)
3891                 return 0;
3892
3893         l_wait_event(mthread->t_ctl_waitq,
3894                      list_empty(&llmd->llmd_req_list) ||
3895                      !thread_is_running(mthread) ||
3896                      thread_is_stopped(athread),
3897                      &lwi);
3898
3899         if (!thread_is_running(mthread) || thread_is_stopped(athread))
3900                 return 0;
3901
3902         down_write(&com->lc_sem);
3903         if (init) {
3904                 lo->ll_pos_latest_start = lfsck->li_pos_current.lp_oit_cookie;
3905         } else {
3906                 lo->ll_pos_last_checkpoint =
3907                                         lfsck->li_pos_current.lp_oit_cookie;
3908                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3909                                 HALF_SEC - lfsck->li_time_last_checkpoint);
3910                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3911                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3912                 com->lc_new_checked = 0;
3913         }
3914
3915         rc = lfsck_layout_store(env, com);
3916         up_write(&com->lc_sem);
3917
3918         return rc;
3919 }
3920
3921 static int lfsck_layout_slave_checkpoint(const struct lu_env *env,
3922                                          struct lfsck_component *com, bool init)
3923 {
3924         struct lfsck_instance   *lfsck = com->lc_lfsck;
3925         struct lfsck_layout     *lo    = com->lc_file_ram;
3926         int                      rc;
3927
3928         if (com->lc_new_checked == 0 && !init)
3929                 return 0;
3930
3931         down_write(&com->lc_sem);
3932
3933         if (init) {
3934                 lo->ll_pos_latest_start = lfsck->li_pos_current.lp_oit_cookie;
3935         } else {
3936                 lo->ll_pos_last_checkpoint =
3937                                         lfsck->li_pos_current.lp_oit_cookie;
3938                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3939                                 HALF_SEC - lfsck->li_time_last_checkpoint);
3940                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3941                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3942                 com->lc_new_checked = 0;
3943         }
3944
3945         rc = lfsck_layout_store(env, com);
3946
3947         up_write(&com->lc_sem);
3948
3949         return rc;
3950 }
3951
3952 static int lfsck_layout_prep(const struct lu_env *env,
3953                              struct lfsck_component *com,
3954                              struct lfsck_start *start)
3955 {
3956         struct lfsck_instance   *lfsck  = com->lc_lfsck;
3957         struct lfsck_layout     *lo     = com->lc_file_ram;
3958         struct lfsck_position   *pos    = &com->lc_pos_start;
3959
3960         fid_zero(&pos->lp_dir_parent);
3961         pos->lp_dir_cookie = 0;
3962         if (lo->ll_status == LS_COMPLETED ||
3963             lo->ll_status == LS_PARTIAL ||
3964             /* To handle orphan, must scan from the beginning. */
3965             (start != NULL && start->ls_flags & LPF_ORPHAN)) {
3966                 int rc;
3967
3968                 rc = lfsck_layout_reset(env, com, false);
3969                 if (rc != 0)
3970                         return rc;
3971         }
3972
3973         down_write(&com->lc_sem);
3974         lo->ll_time_latest_start = cfs_time_current_sec();
3975         spin_lock(&lfsck->li_lock);
3976         if (lo->ll_flags & LF_SCANNED_ONCE) {
3977                 if (!lfsck->li_drop_dryrun ||
3978                     lo->ll_pos_first_inconsistent == 0) {
3979                         lo->ll_status = LS_SCANNING_PHASE2;
3980                         list_del_init(&com->lc_link);
3981                         list_add_tail(&com->lc_link,
3982                                       &lfsck->li_list_double_scan);
3983                         pos->lp_oit_cookie = 0;
3984                 } else {
3985                         int i;
3986
3987                         lo->ll_status = LS_SCANNING_PHASE1;
3988                         lo->ll_run_time_phase1 = 0;
3989                         lo->ll_run_time_phase2 = 0;
3990                         lo->ll_objs_checked_phase1 = 0;
3991                         lo->ll_objs_checked_phase2 = 0;
3992                         lo->ll_objs_failed_phase1 = 0;
3993                         lo->ll_objs_failed_phase2 = 0;
3994                         for (i = 0; i < LLIT_MAX; i++)
3995                                 lo->ll_objs_repaired[i] = 0;
3996
3997                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
3998                         fid_zero(&com->lc_fid_latest_scanned_phase2);
3999                 }
4000         } else {
4001                 lo->ll_status = LS_SCANNING_PHASE1;
4002                 if (!lfsck->li_drop_dryrun ||
4003                     lo->ll_pos_first_inconsistent == 0)
4004                         pos->lp_oit_cookie = lo->ll_pos_last_checkpoint + 1;
4005                 else
4006                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4007         }
4008         spin_unlock(&lfsck->li_lock);
4009         up_write(&com->lc_sem);
4010
4011         return 0;
4012 }
4013
4014 static int lfsck_layout_slave_prep(const struct lu_env *env,
4015                                    struct lfsck_component *com,
4016                                    struct lfsck_start_param *lsp)
4017 {
4018         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4019         struct lfsck_start              *start  = lsp->lsp_start;
4020         int                              rc;
4021
4022         rc = lfsck_layout_prep(env, com, start);
4023         if (rc != 0 || !lsp->lsp_index_valid)
4024                 return rc;
4025
4026         rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
4027         if (rc == 0 && start != NULL && start->ls_flags & LPF_ORPHAN) {
4028                 LASSERT(!llsd->llsd_rbtree_valid);
4029
4030                 write_lock(&llsd->llsd_rb_lock);
4031                 rc = lfsck_rbtree_setup(env, com);
4032                 write_unlock(&llsd->llsd_rb_lock);
4033         }
4034
4035         return rc;
4036 }
4037
4038 static int lfsck_layout_master_prep(const struct lu_env *env,
4039                                     struct lfsck_component *com,
4040                                     struct lfsck_start_param *lsp)
4041 {
4042         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4043         struct lfsck_layout_master_data *llmd    = com->lc_data;
4044         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4045         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4046         struct lfsck_thread_args        *lta;
4047         long                             rc;
4048         ENTRY;
4049
4050         rc = lfsck_layout_prep(env, com, lsp->lsp_start);
4051         if (rc != 0)
4052                 RETURN(rc);
4053
4054         llmd->llmd_assistant_status = 0;
4055         llmd->llmd_post_result = 0;
4056         llmd->llmd_to_post = 0;
4057         llmd->llmd_to_double_scan = 0;
4058         llmd->llmd_in_double_scan = 0;
4059         llmd->llmd_exit = 0;
4060         thread_set_flags(athread, 0);
4061
4062         lta = lfsck_thread_args_init(lfsck, com, lsp);
4063         if (IS_ERR(lta))
4064                 RETURN(PTR_ERR(lta));
4065
4066         rc = PTR_ERR(kthread_run(lfsck_layout_assistant, lta, "lfsck_layout"));
4067         if (IS_ERR_VALUE(rc)) {
4068                 CERROR("%s: Cannot start LFSCK layout assistant thread: "
4069                        "rc = %ld\n", lfsck_lfsck2name(lfsck), rc);
4070                 lfsck_thread_args_fini(lta);
4071         } else {
4072                 struct l_wait_info lwi = { 0 };
4073
4074                 l_wait_event(mthread->t_ctl_waitq,
4075                              thread_is_running(athread) ||
4076                              thread_is_stopped(athread),
4077                              &lwi);
4078                 if (unlikely(!thread_is_running(athread)))
4079                         rc = llmd->llmd_assistant_status;
4080                 else
4081                         rc = 0;
4082         }
4083
4084         RETURN(rc);
4085 }
4086
4087 /* Pre-fetch the attribute for each stripe in the given layout EA. */
4088 static int lfsck_layout_scan_stripes(const struct lu_env *env,
4089                                      struct lfsck_component *com,
4090                                      struct dt_object *parent,
4091                                      struct lov_mds_md_v1 *lmm)
4092 {
4093         struct lfsck_thread_info        *info    = lfsck_env_info(env);
4094         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4095         struct lfsck_bookmark           *bk      = &lfsck->li_bookmark_ram;
4096         struct lfsck_layout             *lo      = com->lc_file_ram;
4097         struct lfsck_layout_master_data *llmd    = com->lc_data;
4098         struct lfsck_layout_object      *llo     = NULL;
4099         struct lov_ost_data_v1          *objs;
4100         struct lfsck_tgt_descs          *ltds    = &lfsck->li_ost_descs;
4101         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4102         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4103                 struct l_wait_info       lwi     = { 0 };
4104         struct lu_buf                   *buf;
4105         int                              rc      = 0;
4106         int                              i;
4107         __u32                            magic;
4108         __u16                            count;
4109         __u16                            gen;
4110         ENTRY;
4111
4112         buf = lfsck_buf_get(env, &info->lti_old_pfid,
4113                             sizeof(struct filter_fid_old));
4114         count = le16_to_cpu(lmm->lmm_stripe_count);
4115         gen = le16_to_cpu(lmm->lmm_layout_gen);
4116         /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
4117          * been verified in lfsck_layout_verify_header() already. If some
4118          * new magic introduced in the future, then layout LFSCK needs to
4119          * be updated also. */
4120         magic = le32_to_cpu(lmm->lmm_magic);
4121         if (magic == LOV_MAGIC_V1) {
4122                 objs = &(lmm->lmm_objects[0]);
4123         } else {
4124                 LASSERT(magic == LOV_MAGIC_V3);
4125                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
4126         }
4127
4128         for (i = 0; i < count; i++, objs++) {
4129                 struct lu_fid           *fid    = &info->lti_fid;
4130                 struct ost_id           *oi     = &info->lti_oi;
4131                 struct lfsck_layout_req *llr;
4132                 struct lfsck_tgt_desc   *tgt    = NULL;
4133                 struct dt_object        *cobj   = NULL;
4134                 __u32                    index  =
4135                                         le32_to_cpu(objs->l_ost_idx);
4136                 bool                     wakeup = false;
4137
4138                 if (is_dummy_lov_ost_data(objs))
4139                         continue;
4140
4141                 l_wait_event(mthread->t_ctl_waitq,
4142                              bk->lb_async_windows == 0 ||
4143                              llmd->llmd_prefetched < bk->lb_async_windows ||
4144                              !thread_is_running(mthread) ||
4145                              thread_is_stopped(athread),
4146                              &lwi);
4147
4148                 if (unlikely(!thread_is_running(mthread)) ||
4149                              thread_is_stopped(athread))
4150                         GOTO(out, rc = 0);
4151
4152                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
4153                 ostid_to_fid(fid, oi, index);
4154                 tgt = lfsck_tgt_get(ltds, index);
4155                 if (unlikely(tgt == NULL)) {
4156                         CERROR("%s: Cannot talk with OST %x which did not join "
4157                                "the layout LFSCK.\n",
4158                                lfsck_lfsck2name(lfsck), index);
4159                         lo->ll_flags |= LF_INCOMPLETE;
4160                         goto next;
4161                 }
4162
4163                 cobj = lfsck_object_find_by_dev(env, tgt->ltd_tgt, fid);
4164                 if (IS_ERR(cobj)) {
4165                         rc = PTR_ERR(cobj);
4166                         goto next;
4167                 }
4168
4169                 rc = dt_declare_attr_get(env, cobj, BYPASS_CAPA);
4170                 if (rc != 0)
4171                         goto next;
4172
4173                 rc = dt_declare_xattr_get(env, cobj, buf, XATTR_NAME_FID,
4174                                           BYPASS_CAPA);
4175                 if (rc != 0)
4176                         goto next;
4177
4178                 if (llo == NULL) {
4179                         llo = lfsck_layout_object_init(env, parent, gen);
4180                         if (IS_ERR(llo)) {
4181                                 rc = PTR_ERR(llo);
4182                                 goto next;
4183                         }
4184                 }
4185
4186                 llr = lfsck_layout_req_init(llo, cobj, index, i);
4187                 if (IS_ERR(llr)) {
4188                         rc = PTR_ERR(llr);
4189                         goto next;
4190                 }
4191
4192                 cobj = NULL;
4193                 spin_lock(&llmd->llmd_lock);
4194                 if (llmd->llmd_assistant_status < 0) {
4195                         spin_unlock(&llmd->llmd_lock);
4196                         lfsck_layout_req_fini(env, llr);
4197                         lfsck_tgt_put(tgt);
4198                         RETURN(llmd->llmd_assistant_status);
4199                 }
4200
4201                 list_add_tail(&llr->llr_list, &llmd->llmd_req_list);
4202                 if (llmd->llmd_prefetched == 0)
4203                         wakeup = true;
4204
4205                 llmd->llmd_prefetched++;
4206                 spin_unlock(&llmd->llmd_lock);
4207                 if (wakeup)
4208                         wake_up_all(&athread->t_ctl_waitq);
4209
4210 next:
4211                 down_write(&com->lc_sem);
4212                 com->lc_new_checked++;
4213                 if (rc < 0)
4214                         lo->ll_objs_failed_phase1++;
4215                 up_write(&com->lc_sem);
4216
4217                 if (cobj != NULL && !IS_ERR(cobj))
4218                         lu_object_put(env, &cobj->do_lu);
4219
4220                 if (likely(tgt != NULL))
4221                         lfsck_tgt_put(tgt);
4222
4223                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
4224                         GOTO(out, rc);
4225         }
4226
4227         GOTO(out, rc = 0);
4228
4229 out:
4230         if (llo != NULL && !IS_ERR(llo))
4231                 lfsck_layout_object_put(env, llo);
4232
4233         return rc;
4234 }
4235
4236 /* For the given object, read its layout EA locally. For each stripe, pre-fetch
4237  * the OST-object's attribute and generate an structure lfsck_layout_req on the
4238  * list ::llmd_req_list.
4239  *
4240  * For each request on above list, the lfsck_layout_assistant thread compares
4241  * the OST side attribute with local attribute, if inconsistent, then repair it.
4242  *
4243  * All above processing is async mode with pipeline. */
4244 static int lfsck_layout_master_exec_oit(const struct lu_env *env,
4245                                         struct lfsck_component *com,
4246                                         struct dt_object *obj)
4247 {
4248         struct lfsck_thread_info        *info   = lfsck_env_info(env);
4249         struct ost_id                   *oi     = &info->lti_oi;
4250         struct lfsck_layout             *lo     = com->lc_file_ram;
4251         struct lfsck_layout_master_data *llmd   = com->lc_data;
4252         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4253         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
4254         struct thandle                  *handle = NULL;
4255         struct lu_buf                   *buf    = &info->lti_big_buf;
4256         struct lov_mds_md_v1            *lmm    = NULL;
4257         struct dt_device                *dev    = lfsck->li_bottom;
4258         struct lustre_handle             lh     = { 0 };
4259         ssize_t                          buflen = buf->lb_len;
4260         int                              rc     = 0;
4261         bool                             locked = false;
4262         bool                             stripe = false;
4263         ENTRY;
4264
4265         if (!S_ISREG(lfsck_object_type(obj)))
4266                 GOTO(out, rc = 0);
4267
4268         if (llmd->llmd_assistant_status < 0)
4269                 GOTO(out, rc = -ESRCH);
4270
4271         fid_to_lmm_oi(lfsck_dto2fid(obj), oi);
4272         lmm_oi_cpu_to_le(oi, oi);
4273         dt_read_lock(env, obj, 0);
4274         locked = true;
4275
4276 again:
4277         rc = lfsck_layout_get_lovea(env, obj, buf, &buflen);
4278         if (rc <= 0)
4279                 GOTO(out, rc);
4280
4281         buf->lb_len = rc;
4282         lmm = buf->lb_buf;
4283         rc = lfsck_layout_verify_header(lmm);
4284         if (rc != 0)
4285                 GOTO(out, rc);
4286
4287         if (memcmp(oi, &lmm->lmm_oi, sizeof(*oi)) == 0)
4288                 GOTO(out, stripe = true);
4289
4290         /* Inconsistent lmm_oi, should be repaired. */
4291         CDEBUG(D_LFSCK, "Repair bad lmm_oi for "DFID"\n",
4292                PFID(lfsck_dto2fid(obj)));
4293
4294         if (bk->lb_param & LPF_DRYRUN) {
4295                 down_write(&com->lc_sem);
4296                 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4297                 up_write(&com->lc_sem);
4298
4299                 GOTO(out, stripe = true);
4300         }
4301
4302         if (!lustre_handle_is_used(&lh)) {
4303                 dt_read_unlock(env, obj);
4304                 locked = false;
4305                 buf->lb_len = buflen;
4306                 rc = lfsck_layout_lock(env, com, obj, &lh,
4307                                        MDS_INODELOCK_LAYOUT |
4308                                        MDS_INODELOCK_XATTR);
4309                 if (rc != 0)
4310                         GOTO(out, rc);
4311
4312                 handle = dt_trans_create(env, dev);
4313                 if (IS_ERR(handle))
4314                         GOTO(out, rc = PTR_ERR(handle));
4315
4316                 rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_LOV,
4317                                           LU_XATTR_REPLACE, handle);
4318                 if (rc != 0)
4319                         GOTO(out, rc);
4320
4321                 rc = dt_trans_start_local(env, dev, handle);
4322                 if (rc != 0)
4323                         GOTO(out, rc);
4324
4325                 dt_write_lock(env, obj, 0);
4326                 locked = true;
4327
4328                 goto again;
4329         }
4330
4331         lmm->lmm_oi = *oi;
4332         rc = dt_xattr_set(env, obj, buf, XATTR_NAME_LOV,
4333                           LU_XATTR_REPLACE, handle, BYPASS_CAPA);
4334         if (rc != 0)
4335                 GOTO(out, rc);
4336
4337         down_write(&com->lc_sem);
4338         lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4339         up_write(&com->lc_sem);
4340
4341         GOTO(out, stripe = true);
4342
4343 out:
4344         if (locked) {
4345                 if (lustre_handle_is_used(&lh))
4346                         dt_write_unlock(env, obj);
4347                 else
4348                         dt_read_unlock(env, obj);
4349         }
4350
4351         if (handle != NULL && !IS_ERR(handle))
4352                 dt_trans_stop(env, dev, handle);
4353
4354         lfsck_layout_unlock(&lh);
4355         if (stripe) {
4356                 rc = lfsck_layout_scan_stripes(env, com, obj, lmm);
4357         } else {
4358                 down_write(&com->lc_sem);
4359                 com->lc_new_checked++;
4360                 if (rc < 0)
4361                         lo->ll_objs_failed_phase1++;
4362                 up_write(&com->lc_sem);
4363         }
4364         buf->lb_len = buflen;
4365
4366         return rc;
4367 }
4368
4369 static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
4370                                        struct lfsck_component *com,
4371                                        struct dt_object *obj)
4372 {
4373         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4374         struct lfsck_layout             *lo     = com->lc_file_ram;
4375         const struct lu_fid             *fid    = lfsck_dto2fid(obj);
4376         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4377         struct lfsck_layout_seq         *lls;
4378         __u64                            seq;
4379         __u64                            oid;
4380         int                              rc;
4381         ENTRY;
4382
4383         LASSERT(llsd != NULL);
4384
4385         lfsck_rbtree_update_bitmap(env, com, fid, false);
4386
4387         down_write(&com->lc_sem);
4388         if (fid_is_idif(fid))
4389                 seq = 0;
4390         else if (!fid_is_norm(fid) ||
4391                  !fid_is_for_ostobj(env, lfsck->li_next, obj, fid))
4392                 GOTO(unlock, rc = 0);
4393         else
4394                 seq = fid_seq(fid);
4395         com->lc_new_checked++;
4396
4397         lls = lfsck_layout_seq_lookup(llsd, seq);
4398         if (lls == NULL) {
4399                 OBD_ALLOC_PTR(lls);
4400                 if (unlikely(lls == NULL))
4401                         GOTO(unlock, rc = -ENOMEM);
4402
4403                 INIT_LIST_HEAD(&lls->lls_list);
4404                 lls->lls_seq = seq;
4405                 rc = lfsck_layout_lastid_load(env, com, lls);
4406                 if (rc != 0) {
4407                         lo->ll_objs_failed_phase1++;
4408                         OBD_FREE_PTR(lls);
4409                         GOTO(unlock, rc);
4410                 }
4411
4412                 lfsck_layout_seq_insert(llsd, lls);
4413         }
4414
4415         if (unlikely(fid_is_last_id(fid)))
4416                 GOTO(unlock, rc = 0);
4417
4418         oid = fid_oid(fid);
4419         if (oid > lls->lls_lastid_known)
4420                 lls->lls_lastid_known = oid;
4421
4422         if (oid > lls->lls_lastid) {
4423                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
4424                         /* OFD may create new objects during LFSCK scanning. */
4425                         rc = lfsck_layout_lastid_reload(env, com, lls);
4426                         if (unlikely(rc != 0))
4427                                 CWARN("%s: failed to reload LAST_ID for "LPX64
4428                                       ": rc = %d\n",
4429                                       lfsck_lfsck2name(com->lc_lfsck),
4430                                       lls->lls_seq, rc);
4431                         if (oid <= lls->lls_lastid)
4432                                 GOTO(unlock, rc = 0);
4433
4434                         LASSERT(lfsck->li_out_notify != NULL);
4435
4436                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4437                                              LE_LASTID_REBUILDING);
4438                         lo->ll_flags |= LF_CRASHED_LASTID;
4439                 }
4440
4441                 lls->lls_lastid = oid;
4442                 lls->lls_dirty = 1;
4443         }
4444
4445         GOTO(unlock, rc = 0);
4446
4447 unlock:
4448         up_write(&com->lc_sem);
4449
4450         return rc;
4451 }
4452
4453 static int lfsck_layout_exec_dir(const struct lu_env *env,
4454                                  struct lfsck_component *com,
4455                                  struct dt_object *obj,
4456                                  struct lu_dirent *ent)
4457 {
4458         return 0;
4459 }
4460
4461 static int lfsck_layout_master_post(const struct lu_env *env,
4462                                     struct lfsck_component *com,
4463                                     int result, bool init)
4464 {
4465         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4466         struct lfsck_layout             *lo      = com->lc_file_ram;
4467         struct lfsck_layout_master_data *llmd    = com->lc_data;
4468         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4469         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4470         struct l_wait_info               lwi     = { 0 };
4471         int                              rc;
4472         ENTRY;
4473
4474
4475         llmd->llmd_post_result = result;
4476         llmd->llmd_to_post = 1;
4477         if (llmd->llmd_post_result <= 0)
4478                 llmd->llmd_exit = 1;
4479
4480         wake_up_all(&athread->t_ctl_waitq);
4481         l_wait_event(mthread->t_ctl_waitq,
4482                      (result > 0 && list_empty(&llmd->llmd_req_list)) ||
4483                      thread_is_stopped(athread),
4484                      &lwi);
4485
4486         if (llmd->llmd_assistant_status < 0)
4487                 result = llmd->llmd_assistant_status;
4488
4489         down_write(&com->lc_sem);
4490         spin_lock(&lfsck->li_lock);
4491         /* When LFSCK failed, there may be some prefetched objects those are
4492          * not been processed yet, we do not know the exactly position, then
4493          * just restart from last check-point next time. */
4494         if (!init && !llmd->llmd_exit)
4495                 lo->ll_pos_last_checkpoint =
4496                                         lfsck->li_pos_current.lp_oit_cookie;
4497
4498         if (result > 0) {
4499                 lo->ll_status = LS_SCANNING_PHASE2;
4500                 lo->ll_flags |= LF_SCANNED_ONCE;
4501                 lo->ll_flags &= ~LF_UPGRADE;
4502                 list_del_init(&com->lc_link);
4503                 list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
4504         } else if (result == 0) {
4505                 lo->ll_status = lfsck->li_status;
4506                 if (lo->ll_status == 0)
4507                         lo->ll_status = LS_STOPPED;
4508                 if (lo->ll_status != LS_PAUSED) {
4509                         list_del_init(&com->lc_link);
4510                         list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4511                 }
4512         } else {
4513                 lo->ll_status = LS_FAILED;
4514                 list_del_init(&com->lc_link);
4515                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4516         }
4517         spin_unlock(&lfsck->li_lock);
4518
4519         if (!init) {
4520                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4521                                 HALF_SEC - lfsck->li_time_last_checkpoint);
4522                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4523                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4524                 com->lc_new_checked = 0;
4525         }
4526
4527         rc = lfsck_layout_store(env, com);
4528         up_write(&com->lc_sem);
4529
4530         RETURN(rc);
4531 }
4532
4533 static int lfsck_layout_slave_post(const struct lu_env *env,
4534                                    struct lfsck_component *com,
4535                                    int result, bool init)
4536 {
4537         struct lfsck_instance   *lfsck = com->lc_lfsck;
4538         struct lfsck_layout     *lo    = com->lc_file_ram;
4539         int                      rc;
4540         bool                     done  = false;
4541
4542         rc = lfsck_layout_lastid_store(env, com);
4543         if (rc != 0)
4544                 result = rc;
4545
4546         LASSERT(lfsck->li_out_notify != NULL);
4547
4548         down_write(&com->lc_sem);
4549
4550         spin_lock(&lfsck->li_lock);
4551         if (!init)
4552                 lo->ll_pos_last_checkpoint =
4553                                         lfsck->li_pos_current.lp_oit_cookie;
4554         if (result > 0) {
4555                 lo->ll_status = LS_SCANNING_PHASE2;
4556                 lo->ll_flags |= LF_SCANNED_ONCE;
4557                 if (lo->ll_flags & LF_CRASHED_LASTID) {
4558                         done = true;
4559                         lo->ll_flags &= ~LF_CRASHED_LASTID;
4560                 }
4561                 lo->ll_flags &= ~LF_UPGRADE;
4562                 list_del_init(&com->lc_link);
4563                 list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
4564         } else if (result == 0) {
4565                 lo->ll_status = lfsck->li_status;
4566                 if (lo->ll_status == 0)
4567                         lo->ll_status = LS_STOPPED;
4568                 if (lo->ll_status != LS_PAUSED) {
4569                         list_del_init(&com->lc_link);
4570                         list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4571                 }
4572         } else {
4573                 lo->ll_status = LS_FAILED;
4574                 list_del_init(&com->lc_link);
4575                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4576         }
4577         spin_unlock(&lfsck->li_lock);
4578
4579         if (done)
4580                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4581                                      LE_LASTID_REBUILT);
4582
4583         if (!init) {
4584                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4585                                 HALF_SEC - lfsck->li_time_last_checkpoint);
4586                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4587                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4588                 com->lc_new_checked = 0;
4589         }
4590
4591         rc = lfsck_layout_store(env, com);
4592
4593         up_write(&com->lc_sem);
4594
4595         lfsck_layout_slave_notify_master(env, com, LE_PHASE1_DONE, result);
4596
4597         if (result <= 0)
4598                 lfsck_rbtree_cleanup(env, com);
4599
4600         return rc;
4601 }
4602
4603 static int lfsck_layout_dump(const struct lu_env *env,
4604                              struct lfsck_component *com, char *buf, int len)
4605 {
4606         struct lfsck_instance   *lfsck = com->lc_lfsck;
4607         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
4608         struct lfsck_layout     *lo    = com->lc_file_ram;
4609         int                      save  = len;
4610         int                      ret   = -ENOSPC;
4611         int                      rc;
4612
4613         down_read(&com->lc_sem);
4614         rc = snprintf(buf, len,
4615                       "name: lfsck_layout\n"
4616                       "magic: %#x\n"
4617                       "version: %d\n"
4618                       "status: %s\n",
4619                       lo->ll_magic,
4620                       bk->lb_version,
4621                       lfsck_status2names(lo->ll_status));
4622         if (rc <= 0)
4623                 goto out;
4624
4625         buf += rc;
4626         len -= rc;
4627         rc = lfsck_bits_dump(&buf, &len, lo->ll_flags, lfsck_flags_names,
4628                              "flags");
4629         if (rc < 0)
4630                 goto out;
4631
4632         rc = lfsck_bits_dump(&buf, &len, bk->lb_param, lfsck_param_names,
4633                              "param");
4634         if (rc < 0)
4635                 goto out;
4636
4637         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_complete,
4638                              "time_since_last_completed");
4639         if (rc < 0)
4640                 goto out;
4641
4642         rc = lfsck_time_dump(&buf, &len, lo->ll_time_latest_start,
4643                              "time_since_latest_start");
4644         if (rc < 0)
4645                 goto out;
4646
4647         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_checkpoint,
4648                              "time_since_last_checkpoint");
4649         if (rc < 0)
4650                 goto out;
4651
4652         rc = snprintf(buf, len,
4653                       "latest_start_position: "LPU64"\n"
4654                       "last_checkpoint_position: "LPU64"\n"
4655                       "first_failure_position: "LPU64"\n",
4656                       lo->ll_pos_latest_start,
4657                       lo->ll_pos_last_checkpoint,
4658                       lo->ll_pos_first_inconsistent);
4659         if (rc <= 0)
4660                 goto out;
4661
4662         buf += rc;
4663         len -= rc;
4664
4665         rc = snprintf(buf, len,
4666                       "success_count: %u\n"
4667                       "repaired_dangling: "LPU64"\n"
4668                       "repaired_unmatched_pair: "LPU64"\n"
4669                       "repaired_multiple_referenced: "LPU64"\n"
4670                       "repaired_orphan: "LPU64"\n"
4671                       "repaired_inconsistent_owner: "LPU64"\n"
4672                       "repaired_others: "LPU64"\n"
4673                       "skipped: "LPU64"\n"
4674                       "failed_phase1: "LPU64"\n"
4675                       "failed_phase2: "LPU64"\n",
4676                       lo->ll_success_count,
4677                       lo->ll_objs_repaired[LLIT_DANGLING - 1],
4678                       lo->ll_objs_repaired[LLIT_UNMATCHED_PAIR - 1],
4679                       lo->ll_objs_repaired[LLIT_MULTIPLE_REFERENCED - 1],
4680                       lo->ll_objs_repaired[LLIT_ORPHAN - 1],
4681                       lo->ll_objs_repaired[LLIT_INCONSISTENT_OWNER - 1],
4682                       lo->ll_objs_repaired[LLIT_OTHERS - 1],
4683                       lo->ll_objs_skipped,
4684                       lo->ll_objs_failed_phase1,
4685                       lo->ll_objs_failed_phase2);
4686         if (rc <= 0)
4687                 goto out;
4688
4689         buf += rc;
4690         len -= rc;
4691
4692         if (lo->ll_status == LS_SCANNING_PHASE1) {
4693                 __u64 pos;
4694                 const struct dt_it_ops *iops;
4695                 cfs_duration_t duration = cfs_time_current() -
4696                                           lfsck->li_time_last_checkpoint;
4697                 __u64 checked = lo->ll_objs_checked_phase1 + com->lc_new_checked;
4698                 __u64 speed = checked;
4699                 __u64 new_checked = com->lc_new_checked * HZ;
4700                 __u32 rtime = lo->ll_run_time_phase1 +
4701                               cfs_duration_sec(duration + HALF_SEC);
4702
4703                 if (duration != 0)
4704                         do_div(new_checked, duration);
4705                 if (rtime != 0)
4706                         do_div(speed, rtime);
4707                 rc = snprintf(buf, len,
4708                               "checked_phase1: "LPU64"\n"
4709                               "checked_phase2: "LPU64"\n"
4710                               "run_time_phase1: %u seconds\n"
4711                               "run_time_phase2: %u seconds\n"
4712                               "average_speed_phase1: "LPU64" items/sec\n"
4713                               "average_speed_phase2: N/A\n"
4714                               "real-time_speed_phase1: "LPU64" items/sec\n"
4715                               "real-time_speed_phase2: N/A\n",
4716                               checked,
4717                               lo->ll_objs_checked_phase2,
4718                               rtime,
4719                               lo->ll_run_time_phase2,
4720                               speed,
4721                               new_checked);
4722                 if (rc <= 0)
4723                         goto out;
4724
4725                 buf += rc;
4726                 len -= rc;
4727
4728                 LASSERT(lfsck->li_di_oit != NULL);
4729
4730                 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
4731
4732                 /* The low layer otable-based iteration position may NOT
4733                  * exactly match the layout-based directory traversal
4734                  * cookie. Generally, it is not a serious issue. But the
4735                  * caller should NOT make assumption on that. */
4736                 pos = iops->store(env, lfsck->li_di_oit);
4737                 if (!lfsck->li_current_oit_processed)
4738                         pos--;
4739                 rc = snprintf(buf, len, "current_position: "LPU64"\n", pos);
4740                 if (rc <= 0)
4741                         goto out;
4742
4743                 buf += rc;
4744                 len -= rc;
4745         } else if (lo->ll_status == LS_SCANNING_PHASE2) {
4746                 cfs_duration_t duration = cfs_time_current() -
4747                                           lfsck->li_time_last_checkpoint;
4748                 __u64 checked = lo->ll_objs_checked_phase1 + com->lc_new_checked;
4749                 __u64 speed = checked;
4750                 __u64 new_checked = com->lc_new_checked * HZ;
4751                 __u32 rtime = lo->ll_run_time_phase1 +
4752                               cfs_duration_sec(duration + HALF_SEC);
4753
4754                 if (duration != 0)
4755                         do_div(new_checked, duration);
4756                 if (rtime != 0)
4757                         do_div(speed, rtime);
4758                 rc = snprintf(buf, len,
4759                               "checked_phase1: "LPU64"\n"
4760                               "checked_phase2: "LPU64"\n"
4761                               "run_time_phase1: %u seconds\n"
4762                               "run_time_phase2: %u seconds\n"
4763                               "average_speed_phase1: "LPU64" items/sec\n"
4764                               "average_speed_phase2: N/A\n"
4765                               "real-time_speed_phase1: "LPU64" items/sec\n"
4766                               "real-time_speed_phase2: N/A\n"
4767                               "current_position: "DFID"\n",
4768                               checked,
4769                               lo->ll_objs_checked_phase2,
4770                               rtime,
4771                               lo->ll_run_time_phase2,
4772                               speed,
4773                               new_checked,
4774                               PFID(&com->lc_fid_latest_scanned_phase2));
4775                 if (rc <= 0)
4776                         goto out;
4777
4778                 buf += rc;
4779                 len -= rc;
4780         } else {
4781                 __u64 speed1 = lo->ll_objs_checked_phase1;
4782                 __u64 speed2 = lo->ll_objs_checked_phase2;
4783
4784                 if (lo->ll_run_time_phase1 != 0)
4785                         do_div(speed1, lo->ll_run_time_phase1);
4786                 if (lo->ll_run_time_phase2 != 0)
4787                         do_div(speed2, lo->ll_run_time_phase2);
4788                 rc = snprintf(buf, len,
4789                               "checked_phase1: "LPU64"\n"
4790                               "checked_phase2: "LPU64"\n"
4791                               "run_time_phase1: %u seconds\n"
4792                               "run_time_phase2: %u seconds\n"
4793                               "average_speed_phase1: "LPU64" items/sec\n"
4794                               "average_speed_phase2: "LPU64" objs/sec\n"
4795                               "real-time_speed_phase1: N/A\n"
4796                               "real-time_speed_phase2: N/A\n"
4797                               "current_position: N/A\n",
4798                               lo->ll_objs_checked_phase1,
4799                               lo->ll_objs_checked_phase2,
4800                               lo->ll_run_time_phase1,
4801                               lo->ll_run_time_phase2,
4802                               speed1,
4803                               speed2);
4804                 if (rc <= 0)
4805                         goto out;
4806
4807                 buf += rc;
4808                 len -= rc;
4809         }
4810         ret = save - len;
4811
4812 out:
4813         up_read(&com->lc_sem);
4814
4815         return ret;
4816 }
4817
4818 static int lfsck_layout_master_double_scan(const struct lu_env *env,
4819                                            struct lfsck_component *com)
4820 {
4821         struct lfsck_layout_master_data *llmd    = com->lc_data;
4822         struct ptlrpc_thread            *mthread = &com->lc_lfsck->li_thread;
4823         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4824         struct lfsck_layout             *lo      = com->lc_file_ram;
4825         struct l_wait_info               lwi     = { 0 };
4826
4827         if (unlikely(lo->ll_status != LS_SCANNING_PHASE2))
4828                 return 0;
4829
4830         llmd->llmd_to_double_scan = 1;
4831         wake_up_all(&athread->t_ctl_waitq);
4832         l_wait_event(mthread->t_ctl_waitq,
4833                      llmd->llmd_in_double_scan ||
4834                      thread_is_stopped(athread),
4835                      &lwi);
4836         if (llmd->llmd_assistant_status < 0)
4837                 return llmd->llmd_assistant_status;
4838
4839         return 0;
4840 }
4841
4842 static int lfsck_layout_slave_double_scan(const struct lu_env *env,
4843                                           struct lfsck_component *com)
4844 {
4845         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4846         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4847         struct lfsck_layout             *lo     = com->lc_file_ram;
4848         struct ptlrpc_thread            *thread = &lfsck->li_thread;
4849         int                              rc;
4850         ENTRY;
4851
4852         if (unlikely(lo->ll_status != LS_SCANNING_PHASE2)) {
4853                 lfsck_rbtree_cleanup(env, com);
4854                 lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE, 0);
4855                 RETURN(0);
4856         }
4857
4858         atomic_inc(&lfsck->li_double_scan_count);
4859
4860         com->lc_new_checked = 0;
4861         com->lc_new_scanned = 0;
4862         com->lc_time_last_checkpoint = cfs_time_current();
4863         com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
4864                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
4865
4866         while (1) {
4867                 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(30),
4868                                                      NULL, NULL);
4869
4870                 rc = lfsck_layout_slave_query_master(env, com);
4871                 if (list_empty(&llsd->llsd_master_list)) {
4872                         if (unlikely(!thread_is_running(thread)))
4873                                 rc = 0;
4874                         else
4875                                 rc = 1;
4876
4877                         GOTO(done, rc);
4878                 }
4879
4880                 if (rc < 0)
4881                         GOTO(done, rc);
4882
4883                 rc = l_wait_event(thread->t_ctl_waitq,
4884                                   !thread_is_running(thread) ||
4885                                   list_empty(&llsd->llsd_master_list),
4886                                   &lwi);
4887                 if (unlikely(!thread_is_running(thread)))
4888                         GOTO(done, rc = 0);
4889
4890                 if (rc == -ETIMEDOUT)
4891                         continue;
4892
4893                 GOTO(done, rc = (rc < 0 ? rc : 1));
4894         }
4895
4896 done:
4897         rc = lfsck_layout_double_scan_result(env, com, rc);
4898
4899         lfsck_rbtree_cleanup(env, com);
4900         lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE, rc);
4901         if (atomic_dec_and_test(&lfsck->li_double_scan_count))
4902                 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
4903
4904         return rc;
4905 }
4906
4907 static void lfsck_layout_master_data_release(const struct lu_env *env,
4908                                              struct lfsck_component *com)
4909 {
4910         struct lfsck_layout_master_data *llmd   = com->lc_data;
4911         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4912         struct lfsck_tgt_descs          *ltds;
4913         struct lfsck_tgt_desc           *ltd;
4914         struct lfsck_tgt_desc           *next;
4915
4916         LASSERT(llmd != NULL);
4917         LASSERT(thread_is_init(&llmd->llmd_thread) ||
4918                 thread_is_stopped(&llmd->llmd_thread));
4919         LASSERT(list_empty(&llmd->llmd_req_list));
4920
4921         com->lc_data = NULL;
4922
4923         ltds = &lfsck->li_ost_descs;
4924         spin_lock(&ltds->ltd_lock);
4925         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase1_list,
4926                                  ltd_layout_phase_list) {
4927                 list_del_init(&ltd->ltd_layout_phase_list);
4928         }
4929         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase2_list,
4930                                  ltd_layout_phase_list) {
4931                 list_del_init(&ltd->ltd_layout_phase_list);
4932         }
4933         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_list,
4934                                  ltd_layout_list) {
4935                 list_del_init(&ltd->ltd_layout_list);
4936         }
4937         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_phase1_list,
4938                                  ltd_layout_phase_list) {
4939                 list_del_init(&ltd->ltd_layout_phase_list);
4940         }
4941         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_phase2_list,
4942                                  ltd_layout_phase_list) {
4943                 list_del_init(&ltd->ltd_layout_phase_list);
4944         }
4945         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_list,
4946                                  ltd_layout_list) {
4947                 list_del_init(&ltd->ltd_layout_list);
4948         }
4949         spin_unlock(&ltds->ltd_lock);
4950
4951         OBD_FREE_PTR(llmd);
4952 }
4953
4954 static void lfsck_layout_slave_data_release(const struct lu_env *env,
4955                                             struct lfsck_component *com)
4956 {
4957         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
4958         struct lfsck_layout_seq          *lls;
4959         struct lfsck_layout_seq          *next;
4960         struct lfsck_layout_slave_target *llst;
4961         struct lfsck_layout_slave_target *tmp;
4962
4963         LASSERT(llsd != NULL);
4964
4965         list_for_each_entry_safe(lls, next, &llsd->llsd_seq_list,
4966                                      lls_list) {
4967                 list_del_init(&lls->lls_list);
4968                 lfsck_object_put(env, lls->lls_lastid_obj);
4969                 OBD_FREE_PTR(lls);
4970         }
4971
4972         list_for_each_entry_safe(llst, tmp, &llsd->llsd_master_list,
4973                                  llst_list) {
4974                 list_del_init(&llst->llst_list);
4975                 OBD_FREE_PTR(llst);
4976         }
4977
4978         lfsck_rbtree_cleanup(env, com);
4979         com->lc_data = NULL;
4980         OBD_FREE_PTR(llsd);
4981 }
4982
4983 static void lfsck_layout_master_quit(const struct lu_env *env,
4984                                      struct lfsck_component *com)
4985 {
4986         struct lfsck_layout_master_data *llmd    = com->lc_data;
4987         struct ptlrpc_thread            *mthread = &com->lc_lfsck->li_thread;
4988         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4989         struct l_wait_info               lwi     = { 0 };
4990
4991         llmd->llmd_exit = 1;
4992         wake_up_all(&athread->t_ctl_waitq);
4993         l_wait_event(mthread->t_ctl_waitq,
4994                      thread_is_init(athread) ||
4995                      thread_is_stopped(athread),
4996                      &lwi);
4997 }
4998
4999 static void lfsck_layout_slave_quit(const struct lu_env *env,
5000                                     struct lfsck_component *com)
5001 {
5002         lfsck_rbtree_cleanup(env, com);
5003 }
5004
5005 static int lfsck_layout_master_in_notify(const struct lu_env *env,
5006                                          struct lfsck_component *com,
5007                                          struct lfsck_request *lr)
5008 {
5009         struct lfsck_instance           *lfsck = com->lc_lfsck;
5010         struct lfsck_layout             *lo    = com->lc_file_ram;
5011         struct lfsck_layout_master_data *llmd  = com->lc_data;
5012         struct lfsck_tgt_descs          *ltds;
5013         struct lfsck_tgt_desc           *ltd;
5014         bool                             fail  = false;
5015         ENTRY;
5016
5017         if (lr->lr_event != LE_PHASE1_DONE &&
5018             lr->lr_event != LE_PHASE2_DONE &&
5019             lr->lr_event != LE_PEER_EXIT)
5020                 RETURN(-EINVAL);
5021
5022         if (lr->lr_flags & LEF_FROM_OST)
5023                 ltds = &lfsck->li_ost_descs;
5024         else
5025                 ltds = &lfsck->li_mdt_descs;
5026         spin_lock(&ltds->ltd_lock);
5027         ltd = LTD_TGT(ltds, lr->lr_index);
5028         if (ltd == NULL) {
5029                 spin_unlock(&ltds->ltd_lock);
5030
5031                 RETURN(-ENODEV);
5032         }
5033
5034         list_del_init(&ltd->ltd_layout_phase_list);
5035         switch (lr->lr_event) {
5036         case LE_PHASE1_DONE:
5037                 if (lr->lr_status <= 0) {
5038                         ltd->ltd_layout_done = 1;
5039                         list_del_init(&ltd->ltd_layout_list);
5040                         CWARN("%s: %s %x failed/stopped at phase1: rc = %d.\n",
5041                               lfsck_lfsck2name(lfsck),
5042                               (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5043                               ltd->ltd_index, lr->lr_status);
5044                         lo->ll_flags |= LF_INCOMPLETE;
5045                         fail = true;
5046                         break;
5047                 }
5048
5049                 if (lr->lr_flags & LEF_FROM_OST) {
5050                         if (list_empty(&ltd->ltd_layout_list))
5051                                 list_add_tail(&ltd->ltd_layout_list,
5052                                               &llmd->llmd_ost_list);
5053                         list_add_tail(&ltd->ltd_layout_phase_list,
5054                                       &llmd->llmd_ost_phase2_list);
5055                 } else {
5056                         if (list_empty(&ltd->ltd_layout_list))
5057                                 list_add_tail(&ltd->ltd_layout_list,
5058                                               &llmd->llmd_mdt_list);
5059                         list_add_tail(&ltd->ltd_layout_phase_list,
5060                                       &llmd->llmd_mdt_phase2_list);
5061                 }
5062                 break;
5063         case LE_PHASE2_DONE:
5064                 ltd->ltd_layout_done = 1;
5065                 list_del_init(&ltd->ltd_layout_list);
5066                 break;
5067         case LE_PEER_EXIT:
5068                 fail = true;
5069                 ltd->ltd_layout_done = 1;
5070                 list_del_init(&ltd->ltd_layout_list);
5071                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT)) {
5072                         CWARN("%s: the peer %s %x exit layout LFSCK.\n",
5073                               lfsck_lfsck2name(lfsck),
5074                               (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5075                               ltd->ltd_index);
5076                         lo->ll_flags |= LF_INCOMPLETE;
5077                 }
5078                 break;
5079         default:
5080                 break;
5081         }
5082         spin_unlock(&ltds->ltd_lock);
5083
5084         if (fail && lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5085                 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5086
5087                 memset(stop, 0, sizeof(*stop));
5088                 stop->ls_status = lr->lr_status;
5089                 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5090                 lfsck_stop(env, lfsck->li_bottom, stop);
5091         } else if (lfsck_layout_master_to_orphan(llmd)) {
5092                 wake_up_all(&llmd->llmd_thread.t_ctl_waitq);
5093         }
5094
5095         RETURN(0);
5096 }
5097
5098 static int lfsck_layout_slave_in_notify(const struct lu_env *env,
5099                                         struct lfsck_component *com,
5100                                         struct lfsck_request *lr)
5101 {
5102         struct lfsck_instance            *lfsck = com->lc_lfsck;
5103         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
5104         struct lfsck_layout_slave_target *llst;
5105         ENTRY;
5106
5107         if (lr->lr_event == LE_FID_ACCESSED) {
5108                 lfsck_rbtree_update_bitmap(env, com, &lr->lr_fid, true);
5109
5110                 RETURN(0);
5111         }
5112
5113         if (lr->lr_event == LE_CONDITIONAL_DESTROY) {
5114                 int rc;
5115
5116                 rc = lfsck_layout_slave_conditional_destroy(env, com, lr);
5117
5118                 RETURN(rc);
5119         }
5120
5121         if (lr->lr_event != LE_PHASE2_DONE && lr->lr_event != LE_PEER_EXIT)
5122                 RETURN(-EINVAL);
5123
5124         llst = lfsck_layout_llst_find_and_del(llsd, lr->lr_index, true);
5125         if (llst == NULL)
5126                 RETURN(-ENODEV);
5127
5128         lfsck_layout_llst_put(llst);
5129         if (list_empty(&llsd->llsd_master_list))
5130                 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5131
5132         if (lr->lr_event == LE_PEER_EXIT &&
5133             lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5134                 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5135
5136                 memset(stop, 0, sizeof(*stop));
5137                 stop->ls_status = lr->lr_status;
5138                 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5139                 lfsck_stop(env, lfsck->li_bottom, stop);
5140         }
5141
5142         RETURN(0);
5143 }
5144
5145 static int lfsck_layout_query(const struct lu_env *env,
5146                               struct lfsck_component *com)
5147 {
5148         struct lfsck_layout *lo = com->lc_file_ram;
5149
5150         return lo->ll_status;
5151 }
5152
5153 static int lfsck_layout_master_stop_notify(const struct lu_env *env,
5154                                            struct lfsck_component *com,
5155                                            struct lfsck_tgt_descs *ltds,
5156                                            struct lfsck_tgt_desc *ltd,
5157                                            struct ptlrpc_request_set *set)
5158 {
5159         struct lfsck_thread_info          *info  = lfsck_env_info(env);
5160         struct lfsck_async_interpret_args *laia  = &info->lti_laia;
5161         struct lfsck_request              *lr    = &info->lti_lr;
5162         struct lfsck_instance             *lfsck = com->lc_lfsck;
5163         int                                rc;
5164
5165         spin_lock(&ltds->ltd_lock);
5166         if (list_empty(&ltd->ltd_layout_list)) {
5167                 LASSERT(list_empty(&ltd->ltd_layout_phase_list));
5168                 spin_unlock(&ltds->ltd_lock);
5169
5170                 return 0;
5171         }
5172
5173         list_del_init(&ltd->ltd_layout_phase_list);
5174         list_del_init(&ltd->ltd_layout_list);
5175         spin_unlock(&ltds->ltd_lock);
5176
5177         memset(lr, 0, sizeof(*lr));
5178         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
5179         lr->lr_event = LE_PEER_EXIT;
5180         lr->lr_active = LT_LAYOUT;
5181         lr->lr_status = LS_CO_PAUSED;
5182         if (ltds == &lfsck->li_ost_descs)
5183                 lr->lr_flags = LEF_TO_OST;
5184
5185         laia->laia_com = com;
5186         laia->laia_ltds = ltds;
5187         atomic_inc(&ltd->ltd_ref);
5188         laia->laia_ltd = ltd;
5189         laia->laia_lr = lr;
5190         laia->laia_shared = 0;
5191
5192         rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
5193                                  lfsck_layout_master_async_interpret,
5194                                  laia, LFSCK_NOTIFY);
5195         if (rc != 0) {
5196                 CERROR("%s: Fail to notify %s %x for co-stop: rc = %d\n",
5197                        lfsck_lfsck2name(lfsck),
5198                        (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5199                        ltd->ltd_index, rc);
5200                 lfsck_tgt_put(ltd);
5201         }
5202
5203         return rc;
5204 }
5205
5206 /* with lfsck::li_lock held */
5207 static int lfsck_layout_slave_join(const struct lu_env *env,
5208                                    struct lfsck_component *com,
5209                                    struct lfsck_start_param *lsp)
5210 {
5211         struct lfsck_instance            *lfsck = com->lc_lfsck;
5212         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
5213         struct lfsck_layout_slave_target *llst;
5214         struct lfsck_start               *start = lsp->lsp_start;
5215         int                               rc    = 0;
5216         ENTRY;
5217
5218         if (!lsp->lsp_index_valid || start == NULL ||
5219             !(start->ls_flags & LPF_ALL_TGT) ||
5220             !(lfsck->li_bookmark_ram.lb_param & LPF_ALL_TGT))
5221                 RETURN(-EALREADY);
5222
5223         spin_unlock(&lfsck->li_lock);
5224         rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
5225         spin_lock(&lfsck->li_lock);
5226         if (rc == 0 && !thread_is_running(&lfsck->li_thread)) {
5227                 spin_unlock(&lfsck->li_lock);
5228                 llst = lfsck_layout_llst_find_and_del(llsd, lsp->lsp_index,
5229                                                       true);
5230                 if (llst != NULL)
5231                         lfsck_layout_llst_put(llst);
5232                 spin_lock(&lfsck->li_lock);
5233                 rc = -EAGAIN;
5234         }
5235
5236         RETURN(rc);
5237 }
5238
5239 static struct lfsck_operations lfsck_layout_master_ops = {
5240         .lfsck_reset            = lfsck_layout_reset,
5241         .lfsck_fail             = lfsck_layout_fail,
5242         .lfsck_checkpoint       = lfsck_layout_master_checkpoint,
5243         .lfsck_prep             = lfsck_layout_master_prep,
5244         .lfsck_exec_oit         = lfsck_layout_master_exec_oit,
5245         .lfsck_exec_dir         = lfsck_layout_exec_dir,
5246         .lfsck_post             = lfsck_layout_master_post,
5247         .lfsck_interpret        = lfsck_layout_master_async_interpret,
5248         .lfsck_dump             = lfsck_layout_dump,
5249         .lfsck_double_scan      = lfsck_layout_master_double_scan,
5250         .lfsck_data_release     = lfsck_layout_master_data_release,
5251         .lfsck_quit             = lfsck_layout_master_quit,
5252         .lfsck_in_notify        = lfsck_layout_master_in_notify,
5253         .lfsck_query            = lfsck_layout_query,
5254         .lfsck_stop_notify      = lfsck_layout_master_stop_notify,
5255 };
5256
5257 static struct lfsck_operations lfsck_layout_slave_ops = {
5258         .lfsck_reset            = lfsck_layout_reset,
5259         .lfsck_fail             = lfsck_layout_fail,
5260         .lfsck_checkpoint       = lfsck_layout_slave_checkpoint,
5261         .lfsck_prep             = lfsck_layout_slave_prep,
5262         .lfsck_exec_oit         = lfsck_layout_slave_exec_oit,
5263         .lfsck_exec_dir         = lfsck_layout_exec_dir,
5264         .lfsck_post             = lfsck_layout_slave_post,
5265         .lfsck_dump             = lfsck_layout_dump,
5266         .lfsck_double_scan      = lfsck_layout_slave_double_scan,
5267         .lfsck_data_release     = lfsck_layout_slave_data_release,
5268         .lfsck_quit             = lfsck_layout_slave_quit,
5269         .lfsck_in_notify        = lfsck_layout_slave_in_notify,
5270         .lfsck_query            = lfsck_layout_query,
5271         .lfsck_join             = lfsck_layout_slave_join,
5272 };
5273
5274 int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
5275 {
5276         struct lfsck_component  *com;
5277         struct lfsck_layout     *lo;
5278         struct dt_object        *root = NULL;
5279         struct dt_object        *obj;
5280         int                      rc;
5281         ENTRY;
5282
5283         OBD_ALLOC_PTR(com);
5284         if (com == NULL)
5285                 RETURN(-ENOMEM);
5286
5287         INIT_LIST_HEAD(&com->lc_link);
5288         INIT_LIST_HEAD(&com->lc_link_dir);
5289         init_rwsem(&com->lc_sem);
5290         atomic_set(&com->lc_ref, 1);
5291         com->lc_lfsck = lfsck;
5292         com->lc_type = LT_LAYOUT;
5293         if (lfsck->li_master) {
5294                 struct lfsck_layout_master_data *llmd;
5295
5296                 com->lc_ops = &lfsck_layout_master_ops;
5297                 OBD_ALLOC_PTR(llmd);
5298                 if (llmd == NULL)
5299                         GOTO(out, rc = -ENOMEM);
5300
5301                 INIT_LIST_HEAD(&llmd->llmd_req_list);
5302                 spin_lock_init(&llmd->llmd_lock);
5303                 INIT_LIST_HEAD(&llmd->llmd_ost_list);
5304                 INIT_LIST_HEAD(&llmd->llmd_ost_phase1_list);
5305                 INIT_LIST_HEAD(&llmd->llmd_ost_phase2_list);
5306                 INIT_LIST_HEAD(&llmd->llmd_mdt_list);
5307                 INIT_LIST_HEAD(&llmd->llmd_mdt_phase1_list);
5308                 INIT_LIST_HEAD(&llmd->llmd_mdt_phase2_list);
5309                 init_waitqueue_head(&llmd->llmd_thread.t_ctl_waitq);
5310                 com->lc_data = llmd;
5311         } else {
5312                 struct lfsck_layout_slave_data *llsd;
5313
5314                 com->lc_ops = &lfsck_layout_slave_ops;
5315                 OBD_ALLOC_PTR(llsd);
5316                 if (llsd == NULL)
5317                         GOTO(out, rc = -ENOMEM);
5318
5319                 INIT_LIST_HEAD(&llsd->llsd_seq_list);
5320                 INIT_LIST_HEAD(&llsd->llsd_master_list);
5321                 spin_lock_init(&llsd->llsd_lock);
5322                 llsd->llsd_rb_root = RB_ROOT;
5323                 rwlock_init(&llsd->llsd_rb_lock);
5324                 com->lc_data = llsd;
5325         }
5326         com->lc_file_size = sizeof(*lo);
5327         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
5328         if (com->lc_file_ram == NULL)
5329                 GOTO(out, rc = -ENOMEM);
5330
5331         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
5332         if (com->lc_file_disk == NULL)
5333                 GOTO(out, rc = -ENOMEM);
5334
5335         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
5336         if (IS_ERR(root))
5337                 GOTO(out, rc = PTR_ERR(root));
5338
5339         if (unlikely(!dt_try_as_dir(env, root)))
5340                 GOTO(out, rc = -ENOTDIR);
5341
5342         obj = local_file_find_or_create(env, lfsck->li_los, root,
5343                                         lfsck_layout_name,
5344                                         S_IFREG | S_IRUGO | S_IWUSR);
5345         if (IS_ERR(obj))
5346                 GOTO(out, rc = PTR_ERR(obj));
5347
5348         com->lc_obj = obj;
5349         rc = lfsck_layout_load(env, com);
5350         if (rc > 0)
5351                 rc = lfsck_layout_reset(env, com, true);
5352         else if (rc == -ENOENT)
5353                 rc = lfsck_layout_init(env, com);
5354
5355         if (rc != 0)
5356                 GOTO(out, rc);
5357
5358         lo = com->lc_file_ram;
5359         switch (lo->ll_status) {
5360         case LS_INIT:
5361         case LS_COMPLETED:
5362         case LS_FAILED:
5363         case LS_STOPPED:
5364         case LS_PARTIAL:
5365                 spin_lock(&lfsck->li_lock);
5366                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
5367                 spin_unlock(&lfsck->li_lock);
5368                 break;
5369         default:
5370                 CERROR("%s: unknown lfsck_layout status: rc = %u\n",
5371                        lfsck_lfsck2name(lfsck), lo->ll_status);
5372                 /* fall through */
5373         case LS_SCANNING_PHASE1:
5374         case LS_SCANNING_PHASE2:
5375                 /* No need to store the status to disk right now.
5376                  * If the system crashed before the status stored,
5377                  * it will be loaded back when next time. */
5378                 lo->ll_status = LS_CRASHED;
5379                 lo->ll_flags |= LF_INCOMPLETE;
5380                 /* fall through */
5381         case LS_PAUSED:
5382         case LS_CRASHED:
5383         case LS_CO_FAILED:
5384         case LS_CO_STOPPED:
5385         case LS_CO_PAUSED:
5386                 spin_lock(&lfsck->li_lock);
5387                 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
5388                 spin_unlock(&lfsck->li_lock);
5389                 break;
5390         }
5391
5392         if (lo->ll_flags & LF_CRASHED_LASTID) {
5393                 LASSERT(lfsck->li_out_notify != NULL);
5394
5395                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
5396                                      LE_LASTID_REBUILDING);
5397         }
5398
5399         GOTO(out, rc = 0);
5400
5401 out:
5402         if (root != NULL && !IS_ERR(root))
5403                 lu_object_put(env, &root->do_lu);
5404
5405         if (rc != 0)
5406                 lfsck_component_cleanup(env, com);
5407
5408         return rc;
5409 }
5410
5411 struct lfsck_orphan_it {
5412         struct lfsck_component           *loi_com;
5413         struct lfsck_rbtree_node         *loi_lrn;
5414         struct lfsck_layout_slave_target *loi_llst;
5415         struct lu_fid                     loi_key;
5416         struct lu_orphan_rec              loi_rec;
5417         __u64                             loi_hash;
5418         unsigned int                      loi_over:1;
5419 };
5420
5421 static int lfsck_fid_match_idx(const struct lu_env *env,
5422                                struct lfsck_instance *lfsck,
5423                                const struct lu_fid *fid, int idx)
5424 {
5425         struct seq_server_site  *ss;
5426         struct lu_server_fld    *sf;
5427         struct lu_seq_range      range  = { 0 };
5428         int                      rc;
5429
5430         /* All abnormal cases will be returned to MDT0. */
5431         if (!fid_is_norm(fid)) {
5432                 if (idx == 0)
5433                         return 1;
5434
5435                 return 0;
5436         }
5437
5438         ss = lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
5439         if (unlikely(ss == NULL))
5440                 return -ENOTCONN;
5441
5442         sf = ss->ss_server_fld;
5443         LASSERT(sf != NULL);
5444
5445         fld_range_set_any(&range);
5446         rc = fld_server_lookup(env, sf, fid_seq(fid), &range);
5447         if (rc != 0)
5448                 return rc;
5449
5450         if (!fld_range_is_mdt(&range))
5451                 return -EINVAL;
5452
5453         if (range.lsr_index == idx)
5454                 return 1;
5455
5456         return 0;
5457 }
5458
5459 static void lfsck_layout_destroy_orphan(const struct lu_env *env,
5460                                         struct dt_device *dev,
5461                                         struct dt_object *obj)
5462 {
5463         struct thandle *handle;
5464         int             rc;
5465         ENTRY;
5466
5467         handle = dt_trans_create(env, dev);
5468         if (IS_ERR(handle))
5469                 RETURN_EXIT;
5470
5471         rc = dt_declare_ref_del(env, obj, handle);
5472         if (rc != 0)
5473                 GOTO(stop, rc);
5474
5475         rc = dt_declare_destroy(env, obj, handle);
5476         if (rc != 0)
5477                 GOTO(stop, rc);
5478
5479         rc = dt_trans_start_local(env, dev, handle);
5480         if (rc != 0)
5481                 GOTO(stop, rc);
5482
5483         dt_write_lock(env, obj, 0);
5484         rc = dt_ref_del(env, obj, handle);
5485         if (rc == 0)
5486                 rc = dt_destroy(env, obj, handle);
5487         dt_write_unlock(env, obj);
5488
5489         GOTO(stop, rc);
5490
5491 stop:
5492         dt_trans_stop(env, dev, handle);
5493
5494         RETURN_EXIT;
5495 }
5496
5497 static int lfsck_orphan_index_lookup(const struct lu_env *env,
5498                                      struct dt_object *dt,
5499                                      struct dt_rec *rec,
5500                                      const struct dt_key *key,
5501                                      struct lustre_capa *capa)
5502 {
5503         return -EOPNOTSUPP;
5504 }
5505
5506 static int lfsck_orphan_index_declare_insert(const struct lu_env *env,
5507                                              struct dt_object *dt,
5508                                              const struct dt_rec *rec,
5509                                              const struct dt_key *key,
5510                                              struct thandle *handle)
5511 {
5512         return -EOPNOTSUPP;
5513 }
5514
5515 static int lfsck_orphan_index_insert(const struct lu_env *env,
5516                                      struct dt_object *dt,
5517                                      const struct dt_rec *rec,
5518                                      const struct dt_key *key,
5519                                      struct thandle *handle,
5520                                      struct lustre_capa *capa,
5521                                      int ignore_quota)
5522 {
5523         return -EOPNOTSUPP;
5524 }
5525
5526 static int lfsck_orphan_index_declare_delete(const struct lu_env *env,
5527                                              struct dt_object *dt,
5528                                              const struct dt_key *key,
5529                                              struct thandle *handle)
5530 {
5531         return -EOPNOTSUPP;
5532 }
5533
5534 static int lfsck_orphan_index_delete(const struct lu_env *env,
5535                                      struct dt_object *dt,
5536                                      const struct dt_key *key,
5537                                      struct thandle *handle,
5538                                      struct lustre_capa *capa)
5539 {
5540         return -EOPNOTSUPP;
5541 }
5542
5543 static struct dt_it *lfsck_orphan_it_init(const struct lu_env *env,
5544                                           struct dt_object *dt,
5545                                           __u32 attr,
5546                                           struct lustre_capa *capa)
5547 {
5548         struct dt_device                *dev    = lu2dt_dev(dt->do_lu.lo_dev);
5549         struct lfsck_instance           *lfsck;
5550         struct lfsck_component          *com    = NULL;
5551         struct lfsck_layout_slave_data  *llsd;
5552         struct lfsck_orphan_it          *it     = NULL;
5553         int                              rc     = 0;
5554         ENTRY;
5555
5556         lfsck = lfsck_instance_find(dev, true, false);
5557         if (unlikely(lfsck == NULL))
5558                 RETURN(ERR_PTR(-ENODEV));
5559
5560         com = lfsck_component_find(lfsck, LT_LAYOUT);
5561         if (unlikely(com == NULL))
5562                 GOTO(out, rc = -ENOENT);
5563
5564         llsd = com->lc_data;
5565         if (!llsd->llsd_rbtree_valid)
5566                 GOTO(out, rc = -ESRCH);
5567
5568         OBD_ALLOC_PTR(it);
5569         if (it == NULL)
5570                 GOTO(out, rc = -ENOMEM);
5571
5572         it->loi_llst = lfsck_layout_llst_find_and_del(llsd, attr, false);
5573         if (it->loi_llst == NULL)
5574                 GOTO(out, rc = -ENODEV);
5575
5576         if (dev->dd_record_fid_accessed) {
5577                 /* The first iteration against the rbtree, scan the whole rbtree
5578                  * to remove the nodes which do NOT need to be handled. */
5579                 write_lock(&llsd->llsd_rb_lock);
5580                 if (dev->dd_record_fid_accessed) {
5581                         struct rb_node                  *node;
5582                         struct rb_node                  *next;
5583                         struct lfsck_rbtree_node        *lrn;
5584
5585                         /* No need to record the fid accessing anymore. */
5586                         dev->dd_record_fid_accessed = 0;
5587
5588                         node = rb_first(&llsd->llsd_rb_root);
5589                         while (node != NULL) {
5590                                 next = rb_next(node);
5591                                 lrn = rb_entry(node, struct lfsck_rbtree_node,
5592                                                lrn_node);
5593                                 if (atomic_read(&lrn->lrn_known_count) <=
5594                                     atomic_read(&lrn->lrn_accessed_count)) {
5595                                         rb_erase(node, &llsd->llsd_rb_root);
5596                                         lfsck_rbtree_free(lrn);
5597                                 }
5598                                 node = next;
5599                         }
5600                 }
5601                 write_unlock(&llsd->llsd_rb_lock);
5602         }
5603
5604         /* read lock the rbtree when init, and unlock when fini */
5605         read_lock(&llsd->llsd_rb_lock);
5606         it->loi_com = com;
5607         com = NULL;
5608
5609         GOTO(out, rc = 0);
5610
5611 out:
5612         if (com != NULL)
5613                 lfsck_component_put(env, com);
5614         lfsck_instance_put(env, lfsck);
5615         if (rc != 0) {
5616                 if (it != NULL)
5617                         OBD_FREE_PTR(it);
5618
5619                 it = (struct lfsck_orphan_it *)ERR_PTR(rc);
5620         }
5621
5622         return (struct dt_it *)it;
5623 }
5624
5625 static void lfsck_orphan_it_fini(const struct lu_env *env,
5626                                  struct dt_it *di)
5627 {
5628         struct lfsck_orphan_it           *it    = (struct lfsck_orphan_it *)di;
5629         struct lfsck_component           *com   = it->loi_com;
5630         struct lfsck_layout_slave_data   *llsd;
5631         struct lfsck_layout_slave_target *llst;
5632
5633         if (com != NULL) {
5634                 llsd = com->lc_data;
5635                 read_unlock(&llsd->llsd_rb_lock);
5636                 llst = it->loi_llst;
5637                 LASSERT(llst != NULL);
5638
5639                 /* Save the key and hash for iterate next. */
5640                 llst->llst_fid = it->loi_key;
5641                 llst->llst_hash = it->loi_hash;
5642                 lfsck_layout_llst_put(llst);
5643                 lfsck_component_put(env, com);
5644         }
5645         OBD_FREE_PTR(it);
5646 }
5647
5648 /**
5649  * \retval       +1: the iteration finished
5650  * \retval        0: on success, not finished
5651  * \retval      -ve: on error
5652  */
5653 static int lfsck_orphan_it_next(const struct lu_env *env,
5654                                 struct dt_it *di)
5655 {
5656         struct lfsck_thread_info        *info   = lfsck_env_info(env);
5657         struct filter_fid_old           *pfid   = &info->lti_old_pfid;
5658         struct lu_attr                  *la     = &info->lti_la;
5659         struct lfsck_orphan_it          *it     = (struct lfsck_orphan_it *)di;
5660         struct lu_fid                   *key    = &it->loi_key;
5661         struct lu_orphan_rec            *rec    = &it->loi_rec;
5662         struct lfsck_component          *com    = it->loi_com;
5663         struct lfsck_instance           *lfsck  = com->lc_lfsck;
5664         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
5665         struct dt_object                *obj;
5666         struct lfsck_rbtree_node        *lrn;
5667         int                              pos;
5668         int                              rc;
5669         __u32                            save;
5670         __u32                            idx    = it->loi_llst->llst_index;
5671         bool                             exact  = false;
5672         ENTRY;
5673
5674         if (it->loi_over)
5675                 RETURN(1);
5676
5677 again0:
5678         lrn = it->loi_lrn;
5679         if (lrn == NULL) {
5680                 lrn = lfsck_rbtree_search(llsd, key, &exact);
5681                 if (lrn == NULL) {
5682                         it->loi_over = 1;
5683                         RETURN(1);
5684                 }
5685
5686                 it->loi_lrn = lrn;
5687                 if (!exact) {
5688                         key->f_seq = lrn->lrn_seq;
5689                         key->f_oid = lrn->lrn_first_oid;
5690                         key->f_ver = 0;
5691                 }
5692         } else {
5693                 key->f_oid++;
5694                 if (unlikely(key->f_oid == 0)) {
5695                         key->f_seq++;
5696                         it->loi_lrn = NULL;
5697                         goto again0;
5698                 }
5699
5700                 if (key->f_oid >=
5701                     lrn->lrn_first_oid + LFSCK_RBTREE_BITMAP_WIDTH) {
5702                         it->loi_lrn = NULL;
5703                         goto again0;
5704                 }
5705         }
5706
5707         if (unlikely(atomic_read(&lrn->lrn_known_count) <=
5708                      atomic_read(&lrn->lrn_accessed_count))) {
5709                 struct rb_node *next = rb_next(&lrn->lrn_node);
5710
5711                 while (next != NULL) {
5712                         lrn = rb_entry(next, struct lfsck_rbtree_node,
5713                                        lrn_node);
5714                         if (atomic_read(&lrn->lrn_known_count) >
5715                             atomic_read(&lrn->lrn_accessed_count))
5716                                 break;
5717                         next = rb_next(next);
5718                 }
5719
5720                 if (next == NULL) {
5721                         it->loi_over = 1;
5722                         RETURN(1);
5723                 }
5724
5725                 it->loi_lrn = lrn;
5726                 key->f_seq = lrn->lrn_seq;
5727                 key->f_oid = lrn->lrn_first_oid;
5728                 key->f_ver = 0;
5729         }
5730
5731         pos = key->f_oid - lrn->lrn_first_oid;
5732
5733 again1:
5734         pos = find_next_bit(lrn->lrn_known_bitmap,
5735                             LFSCK_RBTREE_BITMAP_WIDTH, pos);
5736         if (pos >= LFSCK_RBTREE_BITMAP_WIDTH) {
5737                 key->f_oid = lrn->lrn_first_oid + pos;
5738                 if (unlikely(key->f_oid < lrn->lrn_first_oid)) {
5739                         key->f_seq++;
5740                         key->f_oid = 0;
5741                 }
5742                 it->loi_lrn = NULL;
5743                 goto again0;
5744         }
5745
5746         if (test_bit(pos, lrn->lrn_accessed_bitmap)) {
5747                 pos++;
5748                 goto again1;
5749         }
5750
5751         key->f_oid = lrn->lrn_first_oid + pos;
5752         obj = lfsck_object_find(env, lfsck, key);
5753         if (IS_ERR(obj)) {
5754                 rc = PTR_ERR(obj);
5755                 if (rc == -ENOENT) {
5756                         pos++;
5757                         goto again1;
5758                 }
5759                 RETURN(rc);
5760         }
5761
5762         dt_read_lock(env, obj, 0);
5763         if (!dt_object_exists(obj)) {
5764                 dt_read_unlock(env, obj);
5765                 lfsck_object_put(env, obj);
5766                 pos++;
5767                 goto again1;
5768         }
5769
5770         rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
5771         if (rc != 0)
5772                 GOTO(out, rc);
5773
5774         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, pfid, sizeof(*pfid)),
5775                           XATTR_NAME_FID, BYPASS_CAPA);
5776         if (rc == -ENODATA) {
5777                 /* For the pre-created OST-object, update the bitmap to avoid
5778                  * others LFSCK (second phase) iteration to touch it again. */
5779                 if (la->la_ctime == 0) {
5780                         if (!test_and_set_bit(pos, lrn->lrn_accessed_bitmap))
5781                                 atomic_inc(&lrn->lrn_accessed_count);
5782
5783                         /* For the race between repairing dangling referenced
5784                          * MDT-object and unlink the file, it may left orphan
5785                          * OST-object there. Destroy it now! */
5786                         if (unlikely(!(la->la_mode & S_ISUID))) {
5787                                 dt_read_unlock(env, obj);
5788                                 lfsck_layout_destroy_orphan(env,
5789                                                             lfsck->li_bottom,
5790                                                             obj);
5791                                 lfsck_object_put(env, obj);
5792                                 pos++;
5793                                 goto again1;
5794                         }
5795                 } else if (idx == 0) {
5796                         /* If the orphan OST-object has no parent information,
5797                          * regard it as referenced by the MDT-object on MDT0. */
5798                         fid_zero(&rec->lor_fid);
5799                         rec->lor_uid = la->la_uid;
5800                         rec->lor_gid = la->la_gid;
5801                         GOTO(out, rc = 0);
5802                 }
5803
5804                 dt_read_unlock(env, obj);
5805                 lfsck_object_put(env, obj);
5806                 pos++;
5807                 goto again1;
5808         }
5809
5810         if (rc < 0)
5811                 GOTO(out, rc);
5812
5813         if (rc != sizeof(struct filter_fid) &&
5814             rc != sizeof(struct filter_fid_old))
5815                 GOTO(out, rc = -EINVAL);
5816
5817         fid_le_to_cpu(&rec->lor_fid, &pfid->ff_parent);
5818         /* In fact, the ff_parent::f_ver is not the real parent FID::f_ver,
5819          * instead, it is the OST-object index in its parent MDT-object
5820          * layout EA. */
5821         save = rec->lor_fid.f_ver;
5822         rec->lor_fid.f_ver = 0;
5823         rc = lfsck_fid_match_idx(env, lfsck, &rec->lor_fid, idx);
5824         /* If the orphan OST-object does not claim the MDT, then next.
5825          *
5826          * If we do not know whether it matches or not, then return it
5827          * to the MDT for further check. */
5828         if (rc == 0) {
5829                 dt_read_unlock(env, obj);
5830                 lfsck_object_put(env, obj);
5831                 pos++;
5832                 goto again1;
5833         }
5834
5835         rec->lor_fid.f_ver = save;
5836         rec->lor_uid = la->la_uid;
5837         rec->lor_gid = la->la_gid;
5838
5839         CDEBUG(D_LFSCK, "%s: return orphan "DFID", PFID "DFID", owner %u:%u\n",
5840                lfsck_lfsck2name(com->lc_lfsck), PFID(key), PFID(&rec->lor_fid),
5841                rec->lor_uid, rec->lor_gid);
5842
5843         GOTO(out, rc = 0);
5844
5845 out:
5846         dt_read_unlock(env, obj);
5847         lfsck_object_put(env, obj);
5848         if (rc == 0)
5849                 it->loi_hash++;
5850
5851         return rc;
5852 }
5853
5854 /**
5855  * \retval       +1: locate to the exactly position
5856  * \retval        0: cannot locate to the exactly position,
5857  *                   call next() to move to a valid position.
5858  * \retval      -ve: on error
5859  */
5860 static int lfsck_orphan_it_get(const struct lu_env *env,
5861                                struct dt_it *di,
5862                                const struct dt_key *key)
5863 {
5864         struct lfsck_orphan_it  *it   = (struct lfsck_orphan_it *)di;
5865         int                      rc;
5866
5867         it->loi_key = *(struct lu_fid *)key;
5868         rc = lfsck_orphan_it_next(env, di);
5869         if (rc == 1)
5870                 return 0;
5871
5872         if (rc == 0)
5873                 return 1;
5874
5875         return rc;
5876 }
5877
5878 static void lfsck_orphan_it_put(const struct lu_env *env,
5879                                 struct dt_it *di)
5880 {
5881 }
5882
5883 static struct dt_key *lfsck_orphan_it_key(const struct lu_env *env,
5884                                           const struct dt_it *di)
5885 {
5886         struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5887
5888         return (struct dt_key *)&it->loi_key;
5889 }
5890
5891 static int lfsck_orphan_it_key_size(const struct lu_env *env,
5892                                     const struct dt_it *di)
5893 {
5894         return sizeof(struct lu_fid);
5895 }
5896
5897 static int lfsck_orphan_it_rec(const struct lu_env *env,
5898                                const struct dt_it *di,
5899                                struct dt_rec *rec,
5900                                __u32 attr)
5901 {
5902         struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5903
5904         *(struct lu_orphan_rec *)rec = it->loi_rec;
5905
5906         return 0;
5907 }
5908
5909 static __u64 lfsck_orphan_it_store(const struct lu_env *env,
5910                                    const struct dt_it *di)
5911 {
5912         struct lfsck_orphan_it  *it   = (struct lfsck_orphan_it *)di;
5913
5914         return it->loi_hash;
5915 }
5916
5917 /**
5918  * \retval       +1: locate to the exactly position
5919  * \retval        0: cannot locate to the exactly position,
5920  *                   call next() to move to a valid position.
5921  * \retval      -ve: on error
5922  */
5923 static int lfsck_orphan_it_load(const struct lu_env *env,
5924                                 const struct dt_it *di,
5925                                 __u64 hash)
5926 {
5927         struct lfsck_orphan_it           *it   = (struct lfsck_orphan_it *)di;
5928         struct lfsck_layout_slave_target *llst = it->loi_llst;
5929         int                               rc;
5930
5931         LASSERT(llst != NULL);
5932
5933         if (hash != llst->llst_hash) {
5934                 CWARN("%s: the given hash "LPU64" for orphan iteration does "
5935                       "not match the one when fini "LPU64", to be reset.\n",
5936                       lfsck_lfsck2name(it->loi_com->lc_lfsck), hash,
5937                       llst->llst_hash);
5938                 fid_zero(&llst->llst_fid);
5939                 llst->llst_hash = 0;
5940         }
5941
5942         it->loi_key = llst->llst_fid;
5943         it->loi_hash = llst->llst_hash;
5944         rc = lfsck_orphan_it_next(env, (struct dt_it *)di);
5945         if (rc == 1)
5946                 return 0;
5947
5948         if (rc == 0)
5949                 return 1;
5950
5951         return rc;
5952 }
5953
5954 static int lfsck_orphan_it_key_rec(const struct lu_env *env,
5955                                    const struct dt_it *di,
5956                                    void *key_rec)
5957 {
5958         return 0;
5959 }
5960
5961 const struct dt_index_operations lfsck_orphan_index_ops = {
5962         .dio_lookup             = lfsck_orphan_index_lookup,
5963         .dio_declare_insert     = lfsck_orphan_index_declare_insert,
5964         .dio_insert             = lfsck_orphan_index_insert,
5965         .dio_declare_delete     = lfsck_orphan_index_declare_delete,
5966         .dio_delete             = lfsck_orphan_index_delete,
5967         .dio_it = {
5968                 .init           = lfsck_orphan_it_init,
5969                 .fini           = lfsck_orphan_it_fini,
5970                 .get            = lfsck_orphan_it_get,
5971                 .put            = lfsck_orphan_it_put,
5972                 .next           = lfsck_orphan_it_next,
5973                 .key            = lfsck_orphan_it_key,
5974                 .key_size       = lfsck_orphan_it_key_size,
5975                 .rec            = lfsck_orphan_it_rec,
5976                 .store          = lfsck_orphan_it_store,
5977                 .load           = lfsck_orphan_it_load,
5978                 .key_rec        = lfsck_orphan_it_key_rec,
5979         }
5980 };