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                         if (bk->lb_async_windows != 0 &&
3384                             llmd->llmd_prefetched >= bk->lb_async_windows)
3385                                 wakeup = true;
3386
3387                         llmd->llmd_prefetched--;
3388                         spin_unlock(&llmd->llmd_lock);
3389                         if (wakeup)
3390                                 wake_up_all(&mthread->t_ctl_waitq);
3391
3392                         lfsck_layout_req_fini(env, llr);
3393                         if (rc < 0 && bk->lb_param & LPF_FAILOUT)
3394                                 GOTO(cleanup1, rc);
3395                 }
3396
3397                 /* Wakeup the master engine if it is waiting in checkpoint. */
3398                 wake_up_all(&mthread->t_ctl_waitq);
3399
3400                 l_wait_event(athread->t_ctl_waitq,
3401                              !lfsck_layout_req_empty(llmd) ||
3402                              llmd->llmd_exit ||
3403                              llmd->llmd_to_post ||
3404                              llmd->llmd_to_double_scan,
3405                              &lwi);
3406
3407                 if (unlikely(llmd->llmd_exit))
3408                         GOTO(cleanup1, rc = llmd->llmd_post_result);
3409
3410                 if (!list_empty(&llmd->llmd_req_list))
3411                         continue;
3412
3413                 if (llmd->llmd_to_post) {
3414                         llmd->llmd_to_post = 0;
3415                         LASSERT(llmd->llmd_post_result > 0);
3416
3417                         memset(lr, 0, sizeof(*lr));
3418                         lr->lr_event = LE_PHASE1_DONE;
3419                         lr->lr_status = llmd->llmd_post_result;
3420                         rc = lfsck_layout_master_notify_others(env, com, lr);
3421                         if (rc != 0)
3422                                 CERROR("%s: failed to notify others "
3423                                        "for layout post: rc = %d\n",
3424                                        lfsck_lfsck2name(lfsck), rc);
3425
3426                         /* Wakeup the master engine to go ahead. */
3427                         wake_up_all(&mthread->t_ctl_waitq);
3428                 }
3429
3430                 if (llmd->llmd_to_double_scan) {
3431                         llmd->llmd_to_double_scan = 0;
3432                         atomic_inc(&lfsck->li_double_scan_count);
3433                         llmd->llmd_in_double_scan = 1;
3434                         wake_up_all(&mthread->t_ctl_waitq);
3435
3436                         com->lc_new_checked = 0;
3437                         com->lc_new_scanned = 0;
3438                         com->lc_time_last_checkpoint = cfs_time_current();
3439                         com->lc_time_next_checkpoint =
3440                                 com->lc_time_last_checkpoint +
3441                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
3442
3443                         while (llmd->llmd_in_double_scan) {
3444                                 struct lfsck_tgt_descs  *ltds =
3445                                                         &lfsck->li_ost_descs;
3446                                 struct lfsck_tgt_desc   *ltd;
3447
3448                                 rc = lfsck_layout_master_query_others(env, com);
3449                                 if (lfsck_layout_master_to_orphan(llmd))
3450                                         goto orphan;
3451
3452                                 if (rc < 0)
3453                                         GOTO(cleanup2, rc);
3454
3455                                 /* Pull LFSCK status on related targets once
3456                                  * per 30 seconds if we are not notified. */
3457                                 lwi = LWI_TIMEOUT_INTERVAL(cfs_time_seconds(30),
3458                                                            cfs_time_seconds(1),
3459                                                            NULL, NULL);
3460                                 rc = l_wait_event(athread->t_ctl_waitq,
3461                                         lfsck_layout_master_to_orphan(llmd) ||
3462                                         llmd->llmd_exit ||
3463                                         !thread_is_running(mthread),
3464                                         &lwi);
3465
3466                                 if (unlikely(llmd->llmd_exit ||
3467                                              !thread_is_running(mthread)))
3468                                         GOTO(cleanup2, rc = 0);
3469
3470                                 if (rc == -ETIMEDOUT)
3471                                         continue;
3472
3473                                 if (rc < 0)
3474                                         GOTO(cleanup2, rc);
3475
3476 orphan:
3477                                 spin_lock(&ltds->ltd_lock);
3478                                 while (!list_empty(
3479                                                 &llmd->llmd_ost_phase2_list)) {
3480                                         ltd = list_entry(
3481                                               llmd->llmd_ost_phase2_list.next,
3482                                               struct lfsck_tgt_desc,
3483                                               ltd_layout_phase_list);
3484                                         list_del_init(
3485                                                 &ltd->ltd_layout_phase_list);
3486                                         spin_unlock(&ltds->ltd_lock);
3487
3488                                         if (bk->lb_param & LPF_ALL_TGT) {
3489                                                 rc = lfsck_layout_scan_orphan(
3490                                                                 env, com, ltd);
3491                                                 if (rc != 0 &&
3492                                                     bk->lb_param & LPF_FAILOUT)
3493                                                         GOTO(cleanup2, rc);
3494                                         }
3495
3496                                         if (unlikely(llmd->llmd_exit ||
3497                                                 !thread_is_running(mthread)))
3498                                                 GOTO(cleanup2, rc = 0);
3499
3500                                         spin_lock(&ltds->ltd_lock);
3501                                 }
3502
3503                                 if (list_empty(&llmd->llmd_ost_phase1_list)) {
3504                                         spin_unlock(&ltds->ltd_lock);
3505                                         GOTO(cleanup2, rc = 1);
3506                                 }
3507                                 spin_unlock(&ltds->ltd_lock);
3508                         }
3509                 }
3510         }
3511
3512 cleanup1:
3513         /* Cleanup the unfinished requests. */
3514         spin_lock(&llmd->llmd_lock);
3515         if (rc < 0)
3516                 llmd->llmd_assistant_status = rc;
3517
3518         while (!list_empty(&llmd->llmd_req_list)) {
3519                 llr = list_entry(llmd->llmd_req_list.next,
3520                                  struct lfsck_layout_req,
3521                                  llr_list);
3522                 list_del_init(&llr->llr_list);
3523                 llmd->llmd_prefetched--;
3524                 spin_unlock(&llmd->llmd_lock);
3525                 lfsck_layout_req_fini(env, llr);
3526                 spin_lock(&llmd->llmd_lock);
3527         }
3528         spin_unlock(&llmd->llmd_lock);
3529
3530         LASSERTF(llmd->llmd_prefetched == 0, "unmatched prefeteched objs %d\n",
3531                  llmd->llmd_prefetched);
3532
3533 cleanup2:
3534         memset(lr, 0, sizeof(*lr));
3535         if (rc > 0) {
3536                 lr->lr_event = LE_PHASE2_DONE;
3537                 lr->lr_status = rc;
3538         } else if (rc == 0) {
3539                 if (lfsck->li_flags & LPF_ALL_TGT) {
3540                         lr->lr_event = LE_STOP;
3541                         lr->lr_status = LS_STOPPED;
3542                 } else {
3543                         lr->lr_event = LE_PEER_EXIT;
3544                         switch (lfsck->li_status) {
3545                         case LS_PAUSED:
3546                         case LS_CO_PAUSED:
3547                                 lr->lr_status = LS_CO_PAUSED;
3548                                 break;
3549                         case LS_STOPPED:
3550                         case LS_CO_STOPPED:
3551                                 lr->lr_status = LS_CO_STOPPED;
3552                                 break;
3553                         default:
3554                                 CERROR("%s: unknown status: rc = %d\n",
3555                                        lfsck_lfsck2name(lfsck),
3556                                        lfsck->li_status);
3557                                 lr->lr_status = LS_CO_FAILED;
3558                                 break;
3559                         }
3560                 }
3561         } else {
3562                 if (lfsck->li_flags & LPF_ALL_TGT) {
3563                         lr->lr_event = LE_STOP;
3564                         lr->lr_status = LS_FAILED;
3565                 } else {
3566                         lr->lr_event = LE_PEER_EXIT;
3567                         lr->lr_status = LS_CO_FAILED;
3568                 }
3569         }
3570
3571         rc1 = lfsck_layout_master_notify_others(env, com, lr);
3572         if (rc1 != 0) {
3573                 CERROR("%s: failed to notify others for layout quit: rc = %d\n",
3574                        lfsck_lfsck2name(lfsck), rc1);
3575                 rc = rc1;
3576         }
3577
3578         /* Under force exit case, some requests may be just freed without
3579          * verification, those objects should be re-handled when next run.
3580          * So not update the on-disk tracing file under such case. */
3581         if (!llmd->llmd_exit)
3582                 rc1 = lfsck_layout_double_scan_result(env, com, rc);
3583
3584 fini:
3585         if (llmd->llmd_in_double_scan)
3586                 atomic_dec(&lfsck->li_double_scan_count);
3587
3588         spin_lock(&llmd->llmd_lock);
3589         llmd->llmd_assistant_status = (rc1 != 0 ? rc1 : rc);
3590         thread_set_flags(athread, SVC_STOPPED);
3591         wake_up_all(&mthread->t_ctl_waitq);
3592         spin_unlock(&llmd->llmd_lock);
3593         lfsck_thread_args_fini(lta);
3594
3595         return rc;
3596 }
3597
3598 static int
3599 lfsck_layout_slave_async_interpret(const struct lu_env *env,
3600                                    struct ptlrpc_request *req,
3601                                    void *args, int rc)
3602 {
3603         struct lfsck_layout_slave_async_args *llsaa = args;
3604         struct obd_export                    *exp   = llsaa->llsaa_exp;
3605         struct lfsck_component               *com   = llsaa->llsaa_com;
3606         struct lfsck_layout_slave_target     *llst  = llsaa->llsaa_llst;
3607         struct lfsck_layout_slave_data       *llsd  = com->lc_data;
3608         bool                                  done  = false;
3609
3610         if (rc != 0) {
3611                 /* It is quite probably caused by target crash,
3612                  * to make the LFSCK can go ahead, assume that
3613                  * the target finished the LFSCK prcoessing. */
3614                 done = true;
3615         } else {
3616                 struct lfsck_reply *lr;
3617
3618                 lr = req_capsule_server_get(&req->rq_pill, &RMF_LFSCK_REPLY);
3619                 if (lr->lr_status != LS_SCANNING_PHASE1 &&
3620                     lr->lr_status != LS_SCANNING_PHASE2)
3621                         done = true;
3622         }
3623         if (done)
3624                 lfsck_layout_llst_del(llsd, llst);
3625         lfsck_layout_llst_put(llst);
3626         lfsck_component_put(env, com);
3627         class_export_put(exp);
3628
3629         return 0;
3630 }
3631
3632 static int lfsck_layout_async_query(const struct lu_env *env,
3633                                     struct lfsck_component *com,
3634                                     struct obd_export *exp,
3635                                     struct lfsck_layout_slave_target *llst,
3636                                     struct lfsck_request *lr,
3637                                     struct ptlrpc_request_set *set)
3638 {
3639         struct lfsck_layout_slave_async_args *llsaa;
3640         struct ptlrpc_request                *req;
3641         struct lfsck_request                 *tmp;
3642         int                                   rc;
3643         ENTRY;
3644
3645         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_QUERY);
3646         if (req == NULL)
3647                 RETURN(-ENOMEM);
3648
3649         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_QUERY);
3650         if (rc != 0) {
3651                 ptlrpc_request_free(req);
3652                 RETURN(rc);
3653         }
3654
3655         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3656         *tmp = *lr;
3657         ptlrpc_request_set_replen(req);
3658
3659         llsaa = ptlrpc_req_async_args(req);
3660         llsaa->llsaa_exp = exp;
3661         llsaa->llsaa_com = lfsck_component_get(com);
3662         llsaa->llsaa_llst = llst;
3663         req->rq_interpret_reply = lfsck_layout_slave_async_interpret;
3664         ptlrpc_set_add_req(set, req);
3665
3666         RETURN(0);
3667 }
3668
3669 static int lfsck_layout_async_notify(const struct lu_env *env,
3670                                      struct obd_export *exp,
3671                                      struct lfsck_request *lr,
3672                                      struct ptlrpc_request_set *set)
3673 {
3674         struct ptlrpc_request   *req;
3675         struct lfsck_request    *tmp;
3676         int                      rc;
3677         ENTRY;
3678
3679         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LFSCK_NOTIFY);
3680         if (req == NULL)
3681                 RETURN(-ENOMEM);
3682
3683         rc = ptlrpc_request_pack(req, LUSTRE_OBD_VERSION, LFSCK_NOTIFY);
3684         if (rc != 0) {
3685                 ptlrpc_request_free(req);
3686                 RETURN(rc);
3687         }
3688
3689         tmp = req_capsule_client_get(&req->rq_pill, &RMF_LFSCK_REQUEST);
3690         *tmp = *lr;
3691         ptlrpc_request_set_replen(req);
3692         ptlrpc_set_add_req(set, req);
3693
3694         RETURN(0);
3695 }
3696
3697 static int
3698 lfsck_layout_slave_query_master(const struct lu_env *env,
3699                                 struct lfsck_component *com)
3700 {
3701         struct lfsck_request             *lr    = &lfsck_env_info(env)->lti_lr;
3702         struct lfsck_instance            *lfsck = com->lc_lfsck;
3703         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
3704         struct lfsck_layout_slave_target *llst;
3705         struct obd_export                *exp;
3706         struct ptlrpc_request_set        *set;
3707         int                               rc    = 0;
3708         int                               rc1   = 0;
3709         ENTRY;
3710
3711         set = ptlrpc_prep_set();
3712         if (set == NULL)
3713                 RETURN(-ENOMEM);
3714
3715         memset(lr, 0, sizeof(*lr));
3716         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
3717         lr->lr_event = LE_QUERY;
3718         lr->lr_active = LT_LAYOUT;
3719
3720         llsd->llsd_touch_gen++;
3721         spin_lock(&llsd->llsd_lock);
3722         while (!list_empty(&llsd->llsd_master_list)) {
3723                 llst = list_entry(llsd->llsd_master_list.next,
3724                                   struct lfsck_layout_slave_target,
3725                                   llst_list);
3726                 if (llst->llst_gen == llsd->llsd_touch_gen)
3727                         break;
3728
3729                 llst->llst_gen = llsd->llsd_touch_gen;
3730                 list_del(&llst->llst_list);
3731                 list_add_tail(&llst->llst_list,
3732                               &llsd->llsd_master_list);
3733                 atomic_inc(&llst->llst_ref);
3734                 spin_unlock(&llsd->llsd_lock);
3735
3736                 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3737                                                llst->llst_index);
3738                 if (exp == NULL) {
3739                         lfsck_layout_llst_del(llsd, llst);
3740                         lfsck_layout_llst_put(llst);
3741                         spin_lock(&llsd->llsd_lock);
3742                         continue;
3743                 }
3744
3745                 rc = lfsck_layout_async_query(env, com, exp, llst, lr, set);
3746                 if (rc != 0) {
3747                         CERROR("%s: slave fail to query %s for layout: "
3748                                "rc = %d\n", lfsck_lfsck2name(lfsck),
3749                                exp->exp_obd->obd_name, rc);
3750                         rc1 = rc;
3751                         lfsck_layout_llst_put(llst);
3752                         class_export_put(exp);
3753                 }
3754                 spin_lock(&llsd->llsd_lock);
3755         }
3756         spin_unlock(&llsd->llsd_lock);
3757
3758         rc = ptlrpc_set_wait(set);
3759         ptlrpc_set_destroy(set);
3760
3761         RETURN(rc1 != 0 ? rc1 : rc);
3762 }
3763
3764 static void
3765 lfsck_layout_slave_notify_master(const struct lu_env *env,
3766                                  struct lfsck_component *com,
3767                                  enum lfsck_events event, int result)
3768 {
3769         struct lfsck_instance            *lfsck = com->lc_lfsck;
3770         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
3771         struct lfsck_request             *lr    = &lfsck_env_info(env)->lti_lr;
3772         struct lfsck_layout_slave_target *llst;
3773         struct obd_export                *exp;
3774         struct ptlrpc_request_set        *set;
3775         int                               rc;
3776         ENTRY;
3777
3778         set = ptlrpc_prep_set();
3779         if (set == NULL)
3780                 RETURN_EXIT;
3781
3782         memset(lr, 0, sizeof(*lr));
3783         lr->lr_event = event;
3784         lr->lr_flags = LEF_FROM_OST;
3785         lr->lr_status = result;
3786         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
3787         lr->lr_active = LT_LAYOUT;
3788         llsd->llsd_touch_gen++;
3789         spin_lock(&llsd->llsd_lock);
3790         while (!list_empty(&llsd->llsd_master_list)) {
3791                 llst = list_entry(llsd->llsd_master_list.next,
3792                                   struct lfsck_layout_slave_target,
3793                                   llst_list);
3794                 if (llst->llst_gen == llsd->llsd_touch_gen)
3795                         break;
3796
3797                 llst->llst_gen = llsd->llsd_touch_gen;
3798                 list_del(&llst->llst_list);
3799                 list_add_tail(&llst->llst_list,
3800                               &llsd->llsd_master_list);
3801                 atomic_inc(&llst->llst_ref);
3802                 spin_unlock(&llsd->llsd_lock);
3803
3804                 exp = lustre_find_lwp_by_index(lfsck->li_obd->obd_name,
3805                                                llst->llst_index);
3806                 if (exp == NULL) {
3807                         lfsck_layout_llst_del(llsd, llst);
3808                         lfsck_layout_llst_put(llst);
3809                         spin_lock(&llsd->llsd_lock);
3810                         continue;
3811                 }
3812
3813                 rc = lfsck_layout_async_notify(env, exp, lr, set);
3814                 if (rc != 0)
3815                         CERROR("%s: slave fail to notify %s for layout: "
3816                                "rc = %d\n", lfsck_lfsck2name(lfsck),
3817                                exp->exp_obd->obd_name, rc);
3818                 lfsck_layout_llst_put(llst);
3819                 class_export_put(exp);
3820                 spin_lock(&llsd->llsd_lock);
3821         }
3822         spin_unlock(&llsd->llsd_lock);
3823
3824         ptlrpc_set_wait(set);
3825         ptlrpc_set_destroy(set);
3826
3827         RETURN_EXIT;
3828 }
3829
3830 /* layout APIs */
3831
3832 static int lfsck_layout_reset(const struct lu_env *env,
3833                               struct lfsck_component *com, bool init)
3834 {
3835         struct lfsck_layout     *lo    = com->lc_file_ram;
3836         int                      rc;
3837
3838         down_write(&com->lc_sem);
3839         if (init) {
3840                 memset(lo, 0, com->lc_file_size);
3841         } else {
3842                 __u32 count = lo->ll_success_count;
3843                 __u64 last_time = lo->ll_time_last_complete;
3844
3845                 memset(lo, 0, com->lc_file_size);
3846                 lo->ll_success_count = count;
3847                 lo->ll_time_last_complete = last_time;
3848         }
3849
3850         lo->ll_magic = LFSCK_LAYOUT_MAGIC;
3851         lo->ll_status = LS_INIT;
3852
3853         rc = lfsck_layout_store(env, com);
3854         up_write(&com->lc_sem);
3855
3856         return rc;
3857 }
3858
3859 static void lfsck_layout_fail(const struct lu_env *env,
3860                               struct lfsck_component *com, bool new_checked)
3861 {
3862         struct lfsck_layout *lo = com->lc_file_ram;
3863
3864         down_write(&com->lc_sem);
3865         if (new_checked)
3866                 com->lc_new_checked++;
3867         lo->ll_objs_failed_phase1++;
3868         if (lo->ll_pos_first_inconsistent == 0) {
3869                 struct lfsck_instance *lfsck = com->lc_lfsck;
3870
3871                 lo->ll_pos_first_inconsistent =
3872                         lfsck->li_obj_oit->do_index_ops->dio_it.store(env,
3873                                                         lfsck->li_di_oit);
3874         }
3875         up_write(&com->lc_sem);
3876 }
3877
3878 static int lfsck_layout_master_checkpoint(const struct lu_env *env,
3879                                           struct lfsck_component *com, bool init)
3880 {
3881         struct lfsck_instance           *lfsck   = com->lc_lfsck;
3882         struct lfsck_layout             *lo      = com->lc_file_ram;
3883         struct lfsck_layout_master_data *llmd    = com->lc_data;
3884         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
3885         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
3886         struct l_wait_info               lwi     = { 0 };
3887         int                              rc;
3888
3889         if (com->lc_new_checked == 0 && !init)
3890                 return 0;
3891
3892         l_wait_event(mthread->t_ctl_waitq,
3893                      list_empty(&llmd->llmd_req_list) ||
3894                      !thread_is_running(mthread) ||
3895                      thread_is_stopped(athread),
3896                      &lwi);
3897
3898         if (!thread_is_running(mthread) || thread_is_stopped(athread))
3899                 return 0;
3900
3901         down_write(&com->lc_sem);
3902         if (init) {
3903                 lo->ll_pos_latest_start = lfsck->li_pos_current.lp_oit_cookie;
3904         } else {
3905                 lo->ll_pos_last_checkpoint =
3906                                         lfsck->li_pos_current.lp_oit_cookie;
3907                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3908                                 HALF_SEC - lfsck->li_time_last_checkpoint);
3909                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3910                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3911                 com->lc_new_checked = 0;
3912         }
3913
3914         rc = lfsck_layout_store(env, com);
3915         up_write(&com->lc_sem);
3916
3917         return rc;
3918 }
3919
3920 static int lfsck_layout_slave_checkpoint(const struct lu_env *env,
3921                                          struct lfsck_component *com, bool init)
3922 {
3923         struct lfsck_instance   *lfsck = com->lc_lfsck;
3924         struct lfsck_layout     *lo    = com->lc_file_ram;
3925         int                      rc;
3926
3927         if (com->lc_new_checked == 0 && !init)
3928                 return 0;
3929
3930         down_write(&com->lc_sem);
3931
3932         if (init) {
3933                 lo->ll_pos_latest_start = lfsck->li_pos_current.lp_oit_cookie;
3934         } else {
3935                 lo->ll_pos_last_checkpoint =
3936                                         lfsck->li_pos_current.lp_oit_cookie;
3937                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
3938                                 HALF_SEC - lfsck->li_time_last_checkpoint);
3939                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
3940                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
3941                 com->lc_new_checked = 0;
3942         }
3943
3944         rc = lfsck_layout_store(env, com);
3945
3946         up_write(&com->lc_sem);
3947
3948         return rc;
3949 }
3950
3951 static int lfsck_layout_prep(const struct lu_env *env,
3952                              struct lfsck_component *com,
3953                              struct lfsck_start *start)
3954 {
3955         struct lfsck_instance   *lfsck  = com->lc_lfsck;
3956         struct lfsck_layout     *lo     = com->lc_file_ram;
3957         struct lfsck_position   *pos    = &com->lc_pos_start;
3958
3959         fid_zero(&pos->lp_dir_parent);
3960         pos->lp_dir_cookie = 0;
3961         if (lo->ll_status == LS_COMPLETED ||
3962             lo->ll_status == LS_PARTIAL ||
3963             /* To handle orphan, must scan from the beginning. */
3964             (start != NULL && start->ls_flags & LPF_ORPHAN)) {
3965                 int rc;
3966
3967                 rc = lfsck_layout_reset(env, com, false);
3968                 if (rc != 0)
3969                         return rc;
3970         }
3971
3972         down_write(&com->lc_sem);
3973         lo->ll_time_latest_start = cfs_time_current_sec();
3974         spin_lock(&lfsck->li_lock);
3975         if (lo->ll_flags & LF_SCANNED_ONCE) {
3976                 if (!lfsck->li_drop_dryrun ||
3977                     lo->ll_pos_first_inconsistent == 0) {
3978                         lo->ll_status = LS_SCANNING_PHASE2;
3979                         list_del_init(&com->lc_link);
3980                         list_add_tail(&com->lc_link,
3981                                       &lfsck->li_list_double_scan);
3982                         pos->lp_oit_cookie = 0;
3983                 } else {
3984                         int i;
3985
3986                         lo->ll_status = LS_SCANNING_PHASE1;
3987                         lo->ll_run_time_phase1 = 0;
3988                         lo->ll_run_time_phase2 = 0;
3989                         lo->ll_objs_checked_phase1 = 0;
3990                         lo->ll_objs_checked_phase2 = 0;
3991                         lo->ll_objs_failed_phase1 = 0;
3992                         lo->ll_objs_failed_phase2 = 0;
3993                         for (i = 0; i < LLIT_MAX; i++)
3994                                 lo->ll_objs_repaired[i] = 0;
3995
3996                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
3997                         fid_zero(&com->lc_fid_latest_scanned_phase2);
3998                 }
3999         } else {
4000                 lo->ll_status = LS_SCANNING_PHASE1;
4001                 if (!lfsck->li_drop_dryrun ||
4002                     lo->ll_pos_first_inconsistent == 0)
4003                         pos->lp_oit_cookie = lo->ll_pos_last_checkpoint + 1;
4004                 else
4005                         pos->lp_oit_cookie = lo->ll_pos_first_inconsistent;
4006         }
4007         spin_unlock(&lfsck->li_lock);
4008         up_write(&com->lc_sem);
4009
4010         return 0;
4011 }
4012
4013 static int lfsck_layout_slave_prep(const struct lu_env *env,
4014                                    struct lfsck_component *com,
4015                                    struct lfsck_start_param *lsp)
4016 {
4017         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4018         struct lfsck_start              *start  = lsp->lsp_start;
4019         int                              rc;
4020
4021         rc = lfsck_layout_prep(env, com, start);
4022         if (rc != 0 || !lsp->lsp_index_valid)
4023                 return rc;
4024
4025         rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
4026         if (rc == 0 && start != NULL && start->ls_flags & LPF_ORPHAN) {
4027                 LASSERT(!llsd->llsd_rbtree_valid);
4028
4029                 write_lock(&llsd->llsd_rb_lock);
4030                 rc = lfsck_rbtree_setup(env, com);
4031                 write_unlock(&llsd->llsd_rb_lock);
4032         }
4033
4034         return rc;
4035 }
4036
4037 static int lfsck_layout_master_prep(const struct lu_env *env,
4038                                     struct lfsck_component *com,
4039                                     struct lfsck_start_param *lsp)
4040 {
4041         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4042         struct lfsck_layout_master_data *llmd    = com->lc_data;
4043         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4044         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4045         struct lfsck_thread_args        *lta;
4046         long                             rc;
4047         ENTRY;
4048
4049         rc = lfsck_layout_prep(env, com, lsp->lsp_start);
4050         if (rc != 0)
4051                 RETURN(rc);
4052
4053         llmd->llmd_assistant_status = 0;
4054         llmd->llmd_post_result = 0;
4055         llmd->llmd_to_post = 0;
4056         llmd->llmd_to_double_scan = 0;
4057         llmd->llmd_in_double_scan = 0;
4058         llmd->llmd_exit = 0;
4059         thread_set_flags(athread, 0);
4060
4061         lta = lfsck_thread_args_init(lfsck, com, lsp);
4062         if (IS_ERR(lta))
4063                 RETURN(PTR_ERR(lta));
4064
4065         rc = PTR_ERR(kthread_run(lfsck_layout_assistant, lta, "lfsck_layout"));
4066         if (IS_ERR_VALUE(rc)) {
4067                 CERROR("%s: Cannot start LFSCK layout assistant thread: "
4068                        "rc = %ld\n", lfsck_lfsck2name(lfsck), rc);
4069                 lfsck_thread_args_fini(lta);
4070         } else {
4071                 struct l_wait_info lwi = { 0 };
4072
4073                 l_wait_event(mthread->t_ctl_waitq,
4074                              thread_is_running(athread) ||
4075                              thread_is_stopped(athread),
4076                              &lwi);
4077                 if (unlikely(!thread_is_running(athread)))
4078                         rc = llmd->llmd_assistant_status;
4079                 else
4080                         rc = 0;
4081         }
4082
4083         RETURN(rc);
4084 }
4085
4086 /* Pre-fetch the attribute for each stripe in the given layout EA. */
4087 static int lfsck_layout_scan_stripes(const struct lu_env *env,
4088                                      struct lfsck_component *com,
4089                                      struct dt_object *parent,
4090                                      struct lov_mds_md_v1 *lmm)
4091 {
4092         struct lfsck_thread_info        *info    = lfsck_env_info(env);
4093         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4094         struct lfsck_bookmark           *bk      = &lfsck->li_bookmark_ram;
4095         struct lfsck_layout             *lo      = com->lc_file_ram;
4096         struct lfsck_layout_master_data *llmd    = com->lc_data;
4097         struct lfsck_layout_object      *llo     = NULL;
4098         struct lov_ost_data_v1          *objs;
4099         struct lfsck_tgt_descs          *ltds    = &lfsck->li_ost_descs;
4100         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4101         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4102                 struct l_wait_info       lwi     = { 0 };
4103         struct lu_buf                   *buf;
4104         int                              rc      = 0;
4105         int                              i;
4106         __u32                            magic;
4107         __u16                            count;
4108         __u16                            gen;
4109         ENTRY;
4110
4111         buf = lfsck_buf_get(env, &info->lti_old_pfid,
4112                             sizeof(struct filter_fid_old));
4113         count = le16_to_cpu(lmm->lmm_stripe_count);
4114         gen = le16_to_cpu(lmm->lmm_layout_gen);
4115         /* Currently, we only support LOV_MAGIC_V1/LOV_MAGIC_V3 which has
4116          * been verified in lfsck_layout_verify_header() already. If some
4117          * new magic introduced in the future, then layout LFSCK needs to
4118          * be updated also. */
4119         magic = le32_to_cpu(lmm->lmm_magic);
4120         if (magic == LOV_MAGIC_V1) {
4121                 objs = &(lmm->lmm_objects[0]);
4122         } else {
4123                 LASSERT(magic == LOV_MAGIC_V3);
4124                 objs = &((struct lov_mds_md_v3 *)lmm)->lmm_objects[0];
4125         }
4126
4127         for (i = 0; i < count; i++, objs++) {
4128                 struct lu_fid           *fid    = &info->lti_fid;
4129                 struct ost_id           *oi     = &info->lti_oi;
4130                 struct lfsck_layout_req *llr;
4131                 struct lfsck_tgt_desc   *tgt    = NULL;
4132                 struct dt_object        *cobj   = NULL;
4133                 __u32                    index  =
4134                                         le32_to_cpu(objs->l_ost_idx);
4135                 bool                     wakeup = false;
4136
4137                 if (is_dummy_lov_ost_data(objs))
4138                         continue;
4139
4140                 l_wait_event(mthread->t_ctl_waitq,
4141                              bk->lb_async_windows == 0 ||
4142                              llmd->llmd_prefetched < bk->lb_async_windows ||
4143                              !thread_is_running(mthread) ||
4144                              thread_is_stopped(athread),
4145                              &lwi);
4146
4147                 if (unlikely(!thread_is_running(mthread)) ||
4148                              thread_is_stopped(athread))
4149                         GOTO(out, rc = 0);
4150
4151                 ostid_le_to_cpu(&objs->l_ost_oi, oi);
4152                 ostid_to_fid(fid, oi, index);
4153                 tgt = lfsck_tgt_get(ltds, index);
4154                 if (unlikely(tgt == NULL)) {
4155                         CERROR("%s: Cannot talk with OST %x which did not join "
4156                                "the layout LFSCK.\n",
4157                                lfsck_lfsck2name(lfsck), index);
4158                         lo->ll_flags |= LF_INCOMPLETE;
4159                         goto next;
4160                 }
4161
4162                 cobj = lfsck_object_find_by_dev(env, tgt->ltd_tgt, fid);
4163                 if (IS_ERR(cobj)) {
4164                         rc = PTR_ERR(cobj);
4165                         goto next;
4166                 }
4167
4168                 rc = dt_declare_attr_get(env, cobj, BYPASS_CAPA);
4169                 if (rc != 0)
4170                         goto next;
4171
4172                 rc = dt_declare_xattr_get(env, cobj, buf, XATTR_NAME_FID,
4173                                           BYPASS_CAPA);
4174                 if (rc != 0)
4175                         goto next;
4176
4177                 if (llo == NULL) {
4178                         llo = lfsck_layout_object_init(env, parent, gen);
4179                         if (IS_ERR(llo)) {
4180                                 rc = PTR_ERR(llo);
4181                                 goto next;
4182                         }
4183                 }
4184
4185                 llr = lfsck_layout_req_init(llo, cobj, index, i);
4186                 if (IS_ERR(llr)) {
4187                         rc = PTR_ERR(llr);
4188                         goto next;
4189                 }
4190
4191                 cobj = NULL;
4192                 spin_lock(&llmd->llmd_lock);
4193                 if (llmd->llmd_assistant_status < 0) {
4194                         spin_unlock(&llmd->llmd_lock);
4195                         lfsck_layout_req_fini(env, llr);
4196                         lfsck_tgt_put(tgt);
4197                         RETURN(llmd->llmd_assistant_status);
4198                 }
4199
4200                 list_add_tail(&llr->llr_list, &llmd->llmd_req_list);
4201                 if (llmd->llmd_prefetched == 0)
4202                         wakeup = true;
4203
4204                 llmd->llmd_prefetched++;
4205                 spin_unlock(&llmd->llmd_lock);
4206                 if (wakeup)
4207                         wake_up_all(&athread->t_ctl_waitq);
4208
4209 next:
4210                 down_write(&com->lc_sem);
4211                 com->lc_new_checked++;
4212                 if (rc < 0)
4213                         lo->ll_objs_failed_phase1++;
4214                 up_write(&com->lc_sem);
4215
4216                 if (cobj != NULL && !IS_ERR(cobj))
4217                         lu_object_put(env, &cobj->do_lu);
4218
4219                 if (likely(tgt != NULL))
4220                         lfsck_tgt_put(tgt);
4221
4222                 if (rc < 0 && bk->lb_param & LPF_FAILOUT)
4223                         GOTO(out, rc);
4224         }
4225
4226         GOTO(out, rc = 0);
4227
4228 out:
4229         if (llo != NULL && !IS_ERR(llo))
4230                 lfsck_layout_object_put(env, llo);
4231
4232         return rc;
4233 }
4234
4235 /* For the given object, read its layout EA locally. For each stripe, pre-fetch
4236  * the OST-object's attribute and generate an structure lfsck_layout_req on the
4237  * list ::llmd_req_list.
4238  *
4239  * For each request on above list, the lfsck_layout_assistant thread compares
4240  * the OST side attribute with local attribute, if inconsistent, then repair it.
4241  *
4242  * All above processing is async mode with pipeline. */
4243 static int lfsck_layout_master_exec_oit(const struct lu_env *env,
4244                                         struct lfsck_component *com,
4245                                         struct dt_object *obj)
4246 {
4247         struct lfsck_thread_info        *info   = lfsck_env_info(env);
4248         struct ost_id                   *oi     = &info->lti_oi;
4249         struct lfsck_layout             *lo     = com->lc_file_ram;
4250         struct lfsck_layout_master_data *llmd   = com->lc_data;
4251         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4252         struct lfsck_bookmark           *bk     = &lfsck->li_bookmark_ram;
4253         struct thandle                  *handle = NULL;
4254         struct lu_buf                   *buf    = &info->lti_big_buf;
4255         struct lov_mds_md_v1            *lmm    = NULL;
4256         struct dt_device                *dev    = lfsck->li_bottom;
4257         struct lustre_handle             lh     = { 0 };
4258         ssize_t                          buflen = buf->lb_len;
4259         int                              rc     = 0;
4260         bool                             locked = false;
4261         bool                             stripe = false;
4262         ENTRY;
4263
4264         if (!S_ISREG(lfsck_object_type(obj)))
4265                 GOTO(out, rc = 0);
4266
4267         if (llmd->llmd_assistant_status < 0)
4268                 GOTO(out, rc = -ESRCH);
4269
4270         fid_to_lmm_oi(lfsck_dto2fid(obj), oi);
4271         lmm_oi_cpu_to_le(oi, oi);
4272         dt_read_lock(env, obj, 0);
4273         locked = true;
4274
4275 again:
4276         rc = lfsck_layout_get_lovea(env, obj, buf, &buflen);
4277         if (rc <= 0)
4278                 GOTO(out, rc);
4279
4280         buf->lb_len = rc;
4281         lmm = buf->lb_buf;
4282         rc = lfsck_layout_verify_header(lmm);
4283         if (rc != 0)
4284                 GOTO(out, rc);
4285
4286         if (memcmp(oi, &lmm->lmm_oi, sizeof(*oi)) == 0)
4287                 GOTO(out, stripe = true);
4288
4289         /* Inconsistent lmm_oi, should be repaired. */
4290         CDEBUG(D_LFSCK, "Repair bad lmm_oi for "DFID"\n",
4291                PFID(lfsck_dto2fid(obj)));
4292
4293         if (bk->lb_param & LPF_DRYRUN) {
4294                 down_write(&com->lc_sem);
4295                 lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4296                 up_write(&com->lc_sem);
4297
4298                 GOTO(out, stripe = true);
4299         }
4300
4301         if (!lustre_handle_is_used(&lh)) {
4302                 dt_read_unlock(env, obj);
4303                 locked = false;
4304                 buf->lb_len = buflen;
4305                 rc = lfsck_layout_lock(env, com, obj, &lh,
4306                                        MDS_INODELOCK_LAYOUT |
4307                                        MDS_INODELOCK_XATTR);
4308                 if (rc != 0)
4309                         GOTO(out, rc);
4310
4311                 handle = dt_trans_create(env, dev);
4312                 if (IS_ERR(handle))
4313                         GOTO(out, rc = PTR_ERR(handle));
4314
4315                 rc = dt_declare_xattr_set(env, obj, buf, XATTR_NAME_LOV,
4316                                           LU_XATTR_REPLACE, handle);
4317                 if (rc != 0)
4318                         GOTO(out, rc);
4319
4320                 rc = dt_trans_start_local(env, dev, handle);
4321                 if (rc != 0)
4322                         GOTO(out, rc);
4323
4324                 dt_write_lock(env, obj, 0);
4325                 locked = true;
4326
4327                 goto again;
4328         }
4329
4330         lmm->lmm_oi = *oi;
4331         rc = dt_xattr_set(env, obj, buf, XATTR_NAME_LOV,
4332                           LU_XATTR_REPLACE, handle, BYPASS_CAPA);
4333         if (rc != 0)
4334                 GOTO(out, rc);
4335
4336         down_write(&com->lc_sem);
4337         lo->ll_objs_repaired[LLIT_OTHERS - 1]++;
4338         up_write(&com->lc_sem);
4339
4340         GOTO(out, stripe = true);
4341
4342 out:
4343         if (locked) {
4344                 if (lustre_handle_is_used(&lh))
4345                         dt_write_unlock(env, obj);
4346                 else
4347                         dt_read_unlock(env, obj);
4348         }
4349
4350         if (handle != NULL && !IS_ERR(handle))
4351                 dt_trans_stop(env, dev, handle);
4352
4353         lfsck_layout_unlock(&lh);
4354         if (stripe) {
4355                 rc = lfsck_layout_scan_stripes(env, com, obj, lmm);
4356         } else {
4357                 down_write(&com->lc_sem);
4358                 com->lc_new_checked++;
4359                 if (rc < 0)
4360                         lo->ll_objs_failed_phase1++;
4361                 up_write(&com->lc_sem);
4362         }
4363         buf->lb_len = buflen;
4364
4365         return rc;
4366 }
4367
4368 static int lfsck_layout_slave_exec_oit(const struct lu_env *env,
4369                                        struct lfsck_component *com,
4370                                        struct dt_object *obj)
4371 {
4372         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4373         struct lfsck_layout             *lo     = com->lc_file_ram;
4374         const struct lu_fid             *fid    = lfsck_dto2fid(obj);
4375         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4376         struct lfsck_layout_seq         *lls;
4377         __u64                            seq;
4378         __u64                            oid;
4379         int                              rc;
4380         ENTRY;
4381
4382         LASSERT(llsd != NULL);
4383
4384         lfsck_rbtree_update_bitmap(env, com, fid, false);
4385
4386         down_write(&com->lc_sem);
4387         if (fid_is_idif(fid))
4388                 seq = 0;
4389         else if (!fid_is_norm(fid) ||
4390                  !fid_is_for_ostobj(env, lfsck->li_next, obj, fid))
4391                 GOTO(unlock, rc = 0);
4392         else
4393                 seq = fid_seq(fid);
4394         com->lc_new_checked++;
4395
4396         lls = lfsck_layout_seq_lookup(llsd, seq);
4397         if (lls == NULL) {
4398                 OBD_ALLOC_PTR(lls);
4399                 if (unlikely(lls == NULL))
4400                         GOTO(unlock, rc = -ENOMEM);
4401
4402                 INIT_LIST_HEAD(&lls->lls_list);
4403                 lls->lls_seq = seq;
4404                 rc = lfsck_layout_lastid_load(env, com, lls);
4405                 if (rc != 0) {
4406                         lo->ll_objs_failed_phase1++;
4407                         OBD_FREE_PTR(lls);
4408                         GOTO(unlock, rc);
4409                 }
4410
4411                 lfsck_layout_seq_insert(llsd, lls);
4412         }
4413
4414         if (unlikely(fid_is_last_id(fid)))
4415                 GOTO(unlock, rc = 0);
4416
4417         oid = fid_oid(fid);
4418         if (oid > lls->lls_lastid_known)
4419                 lls->lls_lastid_known = oid;
4420
4421         if (oid > lls->lls_lastid) {
4422                 if (!(lo->ll_flags & LF_CRASHED_LASTID)) {
4423                         /* OFD may create new objects during LFSCK scanning. */
4424                         rc = lfsck_layout_lastid_reload(env, com, lls);
4425                         if (unlikely(rc != 0))
4426                                 CWARN("%s: failed to reload LAST_ID for "LPX64
4427                                       ": rc = %d\n",
4428                                       lfsck_lfsck2name(com->lc_lfsck),
4429                                       lls->lls_seq, rc);
4430                         if (oid <= lls->lls_lastid)
4431                                 GOTO(unlock, rc = 0);
4432
4433                         LASSERT(lfsck->li_out_notify != NULL);
4434
4435                         lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4436                                              LE_LASTID_REBUILDING);
4437                         lo->ll_flags |= LF_CRASHED_LASTID;
4438                 }
4439
4440                 lls->lls_lastid = oid;
4441                 lls->lls_dirty = 1;
4442         }
4443
4444         GOTO(unlock, rc = 0);
4445
4446 unlock:
4447         up_write(&com->lc_sem);
4448
4449         return rc;
4450 }
4451
4452 static int lfsck_layout_exec_dir(const struct lu_env *env,
4453                                  struct lfsck_component *com,
4454                                  struct dt_object *obj,
4455                                  struct lu_dirent *ent)
4456 {
4457         return 0;
4458 }
4459
4460 static int lfsck_layout_master_post(const struct lu_env *env,
4461                                     struct lfsck_component *com,
4462                                     int result, bool init)
4463 {
4464         struct lfsck_instance           *lfsck   = com->lc_lfsck;
4465         struct lfsck_layout             *lo      = com->lc_file_ram;
4466         struct lfsck_layout_master_data *llmd    = com->lc_data;
4467         struct ptlrpc_thread            *mthread = &lfsck->li_thread;
4468         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4469         struct l_wait_info               lwi     = { 0 };
4470         int                              rc;
4471         ENTRY;
4472
4473
4474         llmd->llmd_post_result = result;
4475         llmd->llmd_to_post = 1;
4476         if (llmd->llmd_post_result <= 0)
4477                 llmd->llmd_exit = 1;
4478
4479         wake_up_all(&athread->t_ctl_waitq);
4480         l_wait_event(mthread->t_ctl_waitq,
4481                      (result > 0 && list_empty(&llmd->llmd_req_list)) ||
4482                      thread_is_stopped(athread),
4483                      &lwi);
4484
4485         if (llmd->llmd_assistant_status < 0)
4486                 result = llmd->llmd_assistant_status;
4487
4488         down_write(&com->lc_sem);
4489         spin_lock(&lfsck->li_lock);
4490         /* When LFSCK failed, there may be some prefetched objects those are
4491          * not been processed yet, we do not know the exactly position, then
4492          * just restart from last check-point next time. */
4493         if (!init && !llmd->llmd_exit)
4494                 lo->ll_pos_last_checkpoint =
4495                                         lfsck->li_pos_current.lp_oit_cookie;
4496
4497         if (result > 0) {
4498                 lo->ll_status = LS_SCANNING_PHASE2;
4499                 lo->ll_flags |= LF_SCANNED_ONCE;
4500                 lo->ll_flags &= ~LF_UPGRADE;
4501                 list_del_init(&com->lc_link);
4502                 list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
4503         } else if (result == 0) {
4504                 lo->ll_status = lfsck->li_status;
4505                 if (lo->ll_status == 0)
4506                         lo->ll_status = LS_STOPPED;
4507                 if (lo->ll_status != LS_PAUSED) {
4508                         list_del_init(&com->lc_link);
4509                         list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4510                 }
4511         } else {
4512                 lo->ll_status = LS_FAILED;
4513                 list_del_init(&com->lc_link);
4514                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4515         }
4516         spin_unlock(&lfsck->li_lock);
4517
4518         if (!init) {
4519                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4520                                 HALF_SEC - lfsck->li_time_last_checkpoint);
4521                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4522                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4523                 com->lc_new_checked = 0;
4524         }
4525
4526         rc = lfsck_layout_store(env, com);
4527         up_write(&com->lc_sem);
4528
4529         RETURN(rc);
4530 }
4531
4532 static int lfsck_layout_slave_post(const struct lu_env *env,
4533                                    struct lfsck_component *com,
4534                                    int result, bool init)
4535 {
4536         struct lfsck_instance   *lfsck = com->lc_lfsck;
4537         struct lfsck_layout     *lo    = com->lc_file_ram;
4538         int                      rc;
4539         bool                     done  = false;
4540
4541         rc = lfsck_layout_lastid_store(env, com);
4542         if (rc != 0)
4543                 result = rc;
4544
4545         LASSERT(lfsck->li_out_notify != NULL);
4546
4547         down_write(&com->lc_sem);
4548
4549         spin_lock(&lfsck->li_lock);
4550         if (!init)
4551                 lo->ll_pos_last_checkpoint =
4552                                         lfsck->li_pos_current.lp_oit_cookie;
4553         if (result > 0) {
4554                 lo->ll_status = LS_SCANNING_PHASE2;
4555                 lo->ll_flags |= LF_SCANNED_ONCE;
4556                 if (lo->ll_flags & LF_CRASHED_LASTID) {
4557                         done = true;
4558                         lo->ll_flags &= ~LF_CRASHED_LASTID;
4559                 }
4560                 lo->ll_flags &= ~LF_UPGRADE;
4561                 list_del_init(&com->lc_link);
4562                 list_add_tail(&com->lc_link, &lfsck->li_list_double_scan);
4563         } else if (result == 0) {
4564                 lo->ll_status = lfsck->li_status;
4565                 if (lo->ll_status == 0)
4566                         lo->ll_status = LS_STOPPED;
4567                 if (lo->ll_status != LS_PAUSED) {
4568                         list_del_init(&com->lc_link);
4569                         list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4570                 }
4571         } else {
4572                 lo->ll_status = LS_FAILED;
4573                 list_del_init(&com->lc_link);
4574                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
4575         }
4576         spin_unlock(&lfsck->li_lock);
4577
4578         if (done)
4579                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
4580                                      LE_LASTID_REBUILT);
4581
4582         if (!init) {
4583                 lo->ll_run_time_phase1 += cfs_duration_sec(cfs_time_current() +
4584                                 HALF_SEC - lfsck->li_time_last_checkpoint);
4585                 lo->ll_time_last_checkpoint = cfs_time_current_sec();
4586                 lo->ll_objs_checked_phase1 += com->lc_new_checked;
4587                 com->lc_new_checked = 0;
4588         }
4589
4590         rc = lfsck_layout_store(env, com);
4591
4592         up_write(&com->lc_sem);
4593
4594         lfsck_layout_slave_notify_master(env, com, LE_PHASE1_DONE, result);
4595
4596         if (result <= 0)
4597                 lfsck_rbtree_cleanup(env, com);
4598
4599         return rc;
4600 }
4601
4602 static int lfsck_layout_dump(const struct lu_env *env,
4603                              struct lfsck_component *com, char *buf, int len)
4604 {
4605         struct lfsck_instance   *lfsck = com->lc_lfsck;
4606         struct lfsck_bookmark   *bk    = &lfsck->li_bookmark_ram;
4607         struct lfsck_layout     *lo    = com->lc_file_ram;
4608         int                      save  = len;
4609         int                      ret   = -ENOSPC;
4610         int                      rc;
4611
4612         down_read(&com->lc_sem);
4613         rc = snprintf(buf, len,
4614                       "name: lfsck_layout\n"
4615                       "magic: %#x\n"
4616                       "version: %d\n"
4617                       "status: %s\n",
4618                       lo->ll_magic,
4619                       bk->lb_version,
4620                       lfsck_status2names(lo->ll_status));
4621         if (rc <= 0)
4622                 goto out;
4623
4624         buf += rc;
4625         len -= rc;
4626         rc = lfsck_bits_dump(&buf, &len, lo->ll_flags, lfsck_flags_names,
4627                              "flags");
4628         if (rc < 0)
4629                 goto out;
4630
4631         rc = lfsck_bits_dump(&buf, &len, bk->lb_param, lfsck_param_names,
4632                              "param");
4633         if (rc < 0)
4634                 goto out;
4635
4636         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_complete,
4637                              "time_since_last_completed");
4638         if (rc < 0)
4639                 goto out;
4640
4641         rc = lfsck_time_dump(&buf, &len, lo->ll_time_latest_start,
4642                              "time_since_latest_start");
4643         if (rc < 0)
4644                 goto out;
4645
4646         rc = lfsck_time_dump(&buf, &len, lo->ll_time_last_checkpoint,
4647                              "time_since_last_checkpoint");
4648         if (rc < 0)
4649                 goto out;
4650
4651         rc = snprintf(buf, len,
4652                       "latest_start_position: "LPU64"\n"
4653                       "last_checkpoint_position: "LPU64"\n"
4654                       "first_failure_position: "LPU64"\n",
4655                       lo->ll_pos_latest_start,
4656                       lo->ll_pos_last_checkpoint,
4657                       lo->ll_pos_first_inconsistent);
4658         if (rc <= 0)
4659                 goto out;
4660
4661         buf += rc;
4662         len -= rc;
4663
4664         rc = snprintf(buf, len,
4665                       "success_count: %u\n"
4666                       "repaired_dangling: "LPU64"\n"
4667                       "repaired_unmatched_pair: "LPU64"\n"
4668                       "repaired_multiple_referenced: "LPU64"\n"
4669                       "repaired_orphan: "LPU64"\n"
4670                       "repaired_inconsistent_owner: "LPU64"\n"
4671                       "repaired_others: "LPU64"\n"
4672                       "skipped: "LPU64"\n"
4673                       "failed_phase1: "LPU64"\n"
4674                       "failed_phase2: "LPU64"\n",
4675                       lo->ll_success_count,
4676                       lo->ll_objs_repaired[LLIT_DANGLING - 1],
4677                       lo->ll_objs_repaired[LLIT_UNMATCHED_PAIR - 1],
4678                       lo->ll_objs_repaired[LLIT_MULTIPLE_REFERENCED - 1],
4679                       lo->ll_objs_repaired[LLIT_ORPHAN - 1],
4680                       lo->ll_objs_repaired[LLIT_INCONSISTENT_OWNER - 1],
4681                       lo->ll_objs_repaired[LLIT_OTHERS - 1],
4682                       lo->ll_objs_skipped,
4683                       lo->ll_objs_failed_phase1,
4684                       lo->ll_objs_failed_phase2);
4685         if (rc <= 0)
4686                 goto out;
4687
4688         buf += rc;
4689         len -= rc;
4690
4691         if (lo->ll_status == LS_SCANNING_PHASE1) {
4692                 __u64 pos;
4693                 const struct dt_it_ops *iops;
4694                 cfs_duration_t duration = cfs_time_current() -
4695                                           lfsck->li_time_last_checkpoint;
4696                 __u64 checked = lo->ll_objs_checked_phase1 + com->lc_new_checked;
4697                 __u64 speed = checked;
4698                 __u64 new_checked = com->lc_new_checked * HZ;
4699                 __u32 rtime = lo->ll_run_time_phase1 +
4700                               cfs_duration_sec(duration + HALF_SEC);
4701
4702                 if (duration != 0)
4703                         do_div(new_checked, duration);
4704                 if (rtime != 0)
4705                         do_div(speed, rtime);
4706                 rc = snprintf(buf, len,
4707                               "checked_phase1: "LPU64"\n"
4708                               "checked_phase2: "LPU64"\n"
4709                               "run_time_phase1: %u seconds\n"
4710                               "run_time_phase2: %u seconds\n"
4711                               "average_speed_phase1: "LPU64" items/sec\n"
4712                               "average_speed_phase2: N/A\n"
4713                               "real-time_speed_phase1: "LPU64" items/sec\n"
4714                               "real-time_speed_phase2: N/A\n",
4715                               checked,
4716                               lo->ll_objs_checked_phase2,
4717                               rtime,
4718                               lo->ll_run_time_phase2,
4719                               speed,
4720                               new_checked);
4721                 if (rc <= 0)
4722                         goto out;
4723
4724                 buf += rc;
4725                 len -= rc;
4726
4727                 LASSERT(lfsck->li_di_oit != NULL);
4728
4729                 iops = &lfsck->li_obj_oit->do_index_ops->dio_it;
4730
4731                 /* The low layer otable-based iteration position may NOT
4732                  * exactly match the layout-based directory traversal
4733                  * cookie. Generally, it is not a serious issue. But the
4734                  * caller should NOT make assumption on that. */
4735                 pos = iops->store(env, lfsck->li_di_oit);
4736                 if (!lfsck->li_current_oit_processed)
4737                         pos--;
4738                 rc = snprintf(buf, len, "current_position: "LPU64"\n", pos);
4739                 if (rc <= 0)
4740                         goto out;
4741
4742                 buf += rc;
4743                 len -= rc;
4744         } else if (lo->ll_status == LS_SCANNING_PHASE2) {
4745                 cfs_duration_t duration = cfs_time_current() -
4746                                           lfsck->li_time_last_checkpoint;
4747                 __u64 checked = lo->ll_objs_checked_phase1 + com->lc_new_checked;
4748                 __u64 speed = checked;
4749                 __u64 new_checked = com->lc_new_checked * HZ;
4750                 __u32 rtime = lo->ll_run_time_phase1 +
4751                               cfs_duration_sec(duration + HALF_SEC);
4752
4753                 if (duration != 0)
4754                         do_div(new_checked, duration);
4755                 if (rtime != 0)
4756                         do_div(speed, rtime);
4757                 rc = snprintf(buf, len,
4758                               "checked_phase1: "LPU64"\n"
4759                               "checked_phase2: "LPU64"\n"
4760                               "run_time_phase1: %u seconds\n"
4761                               "run_time_phase2: %u seconds\n"
4762                               "average_speed_phase1: "LPU64" items/sec\n"
4763                               "average_speed_phase2: N/A\n"
4764                               "real-time_speed_phase1: "LPU64" items/sec\n"
4765                               "real-time_speed_phase2: N/A\n"
4766                               "current_position: "DFID"\n",
4767                               checked,
4768                               lo->ll_objs_checked_phase2,
4769                               rtime,
4770                               lo->ll_run_time_phase2,
4771                               speed,
4772                               new_checked,
4773                               PFID(&com->lc_fid_latest_scanned_phase2));
4774                 if (rc <= 0)
4775                         goto out;
4776
4777                 buf += rc;
4778                 len -= rc;
4779         } else {
4780                 __u64 speed1 = lo->ll_objs_checked_phase1;
4781                 __u64 speed2 = lo->ll_objs_checked_phase2;
4782
4783                 if (lo->ll_run_time_phase1 != 0)
4784                         do_div(speed1, lo->ll_run_time_phase1);
4785                 if (lo->ll_run_time_phase2 != 0)
4786                         do_div(speed2, lo->ll_run_time_phase2);
4787                 rc = snprintf(buf, len,
4788                               "checked_phase1: "LPU64"\n"
4789                               "checked_phase2: "LPU64"\n"
4790                               "run_time_phase1: %u seconds\n"
4791                               "run_time_phase2: %u seconds\n"
4792                               "average_speed_phase1: "LPU64" items/sec\n"
4793                               "average_speed_phase2: "LPU64" objs/sec\n"
4794                               "real-time_speed_phase1: N/A\n"
4795                               "real-time_speed_phase2: N/A\n"
4796                               "current_position: N/A\n",
4797                               lo->ll_objs_checked_phase1,
4798                               lo->ll_objs_checked_phase2,
4799                               lo->ll_run_time_phase1,
4800                               lo->ll_run_time_phase2,
4801                               speed1,
4802                               speed2);
4803                 if (rc <= 0)
4804                         goto out;
4805
4806                 buf += rc;
4807                 len -= rc;
4808         }
4809         ret = save - len;
4810
4811 out:
4812         up_read(&com->lc_sem);
4813
4814         return ret;
4815 }
4816
4817 static int lfsck_layout_master_double_scan(const struct lu_env *env,
4818                                            struct lfsck_component *com)
4819 {
4820         struct lfsck_layout_master_data *llmd    = com->lc_data;
4821         struct ptlrpc_thread            *mthread = &com->lc_lfsck->li_thread;
4822         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4823         struct lfsck_layout             *lo      = com->lc_file_ram;
4824         struct l_wait_info               lwi     = { 0 };
4825
4826         if (unlikely(lo->ll_status != LS_SCANNING_PHASE2))
4827                 return 0;
4828
4829         llmd->llmd_to_double_scan = 1;
4830         wake_up_all(&athread->t_ctl_waitq);
4831         l_wait_event(mthread->t_ctl_waitq,
4832                      llmd->llmd_in_double_scan ||
4833                      thread_is_stopped(athread),
4834                      &lwi);
4835         if (llmd->llmd_assistant_status < 0)
4836                 return llmd->llmd_assistant_status;
4837
4838         return 0;
4839 }
4840
4841 static int lfsck_layout_slave_double_scan(const struct lu_env *env,
4842                                           struct lfsck_component *com)
4843 {
4844         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4845         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
4846         struct lfsck_layout             *lo     = com->lc_file_ram;
4847         struct ptlrpc_thread            *thread = &lfsck->li_thread;
4848         int                              rc;
4849         ENTRY;
4850
4851         if (unlikely(lo->ll_status != LS_SCANNING_PHASE2)) {
4852                 lfsck_rbtree_cleanup(env, com);
4853                 lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE, 0);
4854                 RETURN(0);
4855         }
4856
4857         atomic_inc(&lfsck->li_double_scan_count);
4858
4859         com->lc_new_checked = 0;
4860         com->lc_new_scanned = 0;
4861         com->lc_time_last_checkpoint = cfs_time_current();
4862         com->lc_time_next_checkpoint = com->lc_time_last_checkpoint +
4863                                 cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
4864
4865         while (1) {
4866                 struct l_wait_info lwi = LWI_TIMEOUT(cfs_time_seconds(30),
4867                                                      NULL, NULL);
4868
4869                 rc = lfsck_layout_slave_query_master(env, com);
4870                 if (list_empty(&llsd->llsd_master_list)) {
4871                         if (unlikely(!thread_is_running(thread)))
4872                                 rc = 0;
4873                         else
4874                                 rc = 1;
4875
4876                         GOTO(done, rc);
4877                 }
4878
4879                 if (rc < 0)
4880                         GOTO(done, rc);
4881
4882                 rc = l_wait_event(thread->t_ctl_waitq,
4883                                   !thread_is_running(thread) ||
4884                                   list_empty(&llsd->llsd_master_list),
4885                                   &lwi);
4886                 if (unlikely(!thread_is_running(thread)))
4887                         GOTO(done, rc = 0);
4888
4889                 if (rc == -ETIMEDOUT)
4890                         continue;
4891
4892                 GOTO(done, rc = (rc < 0 ? rc : 1));
4893         }
4894
4895 done:
4896         rc = lfsck_layout_double_scan_result(env, com, rc);
4897
4898         lfsck_rbtree_cleanup(env, com);
4899         lfsck_layout_slave_notify_master(env, com, LE_PHASE2_DONE, rc);
4900         if (atomic_dec_and_test(&lfsck->li_double_scan_count))
4901                 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
4902
4903         return rc;
4904 }
4905
4906 static void lfsck_layout_master_data_release(const struct lu_env *env,
4907                                              struct lfsck_component *com)
4908 {
4909         struct lfsck_layout_master_data *llmd   = com->lc_data;
4910         struct lfsck_instance           *lfsck  = com->lc_lfsck;
4911         struct lfsck_tgt_descs          *ltds;
4912         struct lfsck_tgt_desc           *ltd;
4913         struct lfsck_tgt_desc           *next;
4914
4915         LASSERT(llmd != NULL);
4916         LASSERT(thread_is_init(&llmd->llmd_thread) ||
4917                 thread_is_stopped(&llmd->llmd_thread));
4918         LASSERT(list_empty(&llmd->llmd_req_list));
4919
4920         com->lc_data = NULL;
4921
4922         ltds = &lfsck->li_ost_descs;
4923         spin_lock(&ltds->ltd_lock);
4924         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase1_list,
4925                                  ltd_layout_phase_list) {
4926                 list_del_init(&ltd->ltd_layout_phase_list);
4927         }
4928         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_phase2_list,
4929                                  ltd_layout_phase_list) {
4930                 list_del_init(&ltd->ltd_layout_phase_list);
4931         }
4932         list_for_each_entry_safe(ltd, next, &llmd->llmd_ost_list,
4933                                  ltd_layout_list) {
4934                 list_del_init(&ltd->ltd_layout_list);
4935         }
4936         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_phase1_list,
4937                                  ltd_layout_phase_list) {
4938                 list_del_init(&ltd->ltd_layout_phase_list);
4939         }
4940         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_phase2_list,
4941                                  ltd_layout_phase_list) {
4942                 list_del_init(&ltd->ltd_layout_phase_list);
4943         }
4944         list_for_each_entry_safe(ltd, next, &llmd->llmd_mdt_list,
4945                                  ltd_layout_list) {
4946                 list_del_init(&ltd->ltd_layout_list);
4947         }
4948         spin_unlock(&ltds->ltd_lock);
4949
4950         OBD_FREE_PTR(llmd);
4951 }
4952
4953 static void lfsck_layout_slave_data_release(const struct lu_env *env,
4954                                             struct lfsck_component *com)
4955 {
4956         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
4957         struct lfsck_layout_seq          *lls;
4958         struct lfsck_layout_seq          *next;
4959         struct lfsck_layout_slave_target *llst;
4960         struct lfsck_layout_slave_target *tmp;
4961
4962         LASSERT(llsd != NULL);
4963
4964         list_for_each_entry_safe(lls, next, &llsd->llsd_seq_list,
4965                                      lls_list) {
4966                 list_del_init(&lls->lls_list);
4967                 lfsck_object_put(env, lls->lls_lastid_obj);
4968                 OBD_FREE_PTR(lls);
4969         }
4970
4971         list_for_each_entry_safe(llst, tmp, &llsd->llsd_master_list,
4972                                  llst_list) {
4973                 list_del_init(&llst->llst_list);
4974                 OBD_FREE_PTR(llst);
4975         }
4976
4977         lfsck_rbtree_cleanup(env, com);
4978         com->lc_data = NULL;
4979         OBD_FREE_PTR(llsd);
4980 }
4981
4982 static void lfsck_layout_master_quit(const struct lu_env *env,
4983                                      struct lfsck_component *com)
4984 {
4985         struct lfsck_layout_master_data *llmd    = com->lc_data;
4986         struct ptlrpc_thread            *mthread = &com->lc_lfsck->li_thread;
4987         struct ptlrpc_thread            *athread = &llmd->llmd_thread;
4988         struct l_wait_info               lwi     = { 0 };
4989
4990         llmd->llmd_exit = 1;
4991         wake_up_all(&athread->t_ctl_waitq);
4992         l_wait_event(mthread->t_ctl_waitq,
4993                      thread_is_init(athread) ||
4994                      thread_is_stopped(athread),
4995                      &lwi);
4996 }
4997
4998 static void lfsck_layout_slave_quit(const struct lu_env *env,
4999                                     struct lfsck_component *com)
5000 {
5001         lfsck_rbtree_cleanup(env, com);
5002 }
5003
5004 static int lfsck_layout_master_in_notify(const struct lu_env *env,
5005                                          struct lfsck_component *com,
5006                                          struct lfsck_request *lr)
5007 {
5008         struct lfsck_instance           *lfsck = com->lc_lfsck;
5009         struct lfsck_layout             *lo    = com->lc_file_ram;
5010         struct lfsck_layout_master_data *llmd  = com->lc_data;
5011         struct lfsck_tgt_descs          *ltds;
5012         struct lfsck_tgt_desc           *ltd;
5013         bool                             fail  = false;
5014         ENTRY;
5015
5016         if (lr->lr_event != LE_PHASE1_DONE &&
5017             lr->lr_event != LE_PHASE2_DONE &&
5018             lr->lr_event != LE_PEER_EXIT)
5019                 RETURN(-EINVAL);
5020
5021         if (lr->lr_flags & LEF_FROM_OST)
5022                 ltds = &lfsck->li_ost_descs;
5023         else
5024                 ltds = &lfsck->li_mdt_descs;
5025         spin_lock(&ltds->ltd_lock);
5026         ltd = LTD_TGT(ltds, lr->lr_index);
5027         if (ltd == NULL) {
5028                 spin_unlock(&ltds->ltd_lock);
5029
5030                 RETURN(-ENODEV);
5031         }
5032
5033         list_del_init(&ltd->ltd_layout_phase_list);
5034         switch (lr->lr_event) {
5035         case LE_PHASE1_DONE:
5036                 if (lr->lr_status <= 0) {
5037                         ltd->ltd_layout_done = 1;
5038                         list_del_init(&ltd->ltd_layout_list);
5039                         CWARN("%s: %s %x failed/stopped at phase1: rc = %d.\n",
5040                               lfsck_lfsck2name(lfsck),
5041                               (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5042                               ltd->ltd_index, lr->lr_status);
5043                         lo->ll_flags |= LF_INCOMPLETE;
5044                         fail = true;
5045                         break;
5046                 }
5047
5048                 if (lr->lr_flags & LEF_FROM_OST) {
5049                         if (list_empty(&ltd->ltd_layout_list))
5050                                 list_add_tail(&ltd->ltd_layout_list,
5051                                               &llmd->llmd_ost_list);
5052                         list_add_tail(&ltd->ltd_layout_phase_list,
5053                                       &llmd->llmd_ost_phase2_list);
5054                 } else {
5055                         if (list_empty(&ltd->ltd_layout_list))
5056                                 list_add_tail(&ltd->ltd_layout_list,
5057                                               &llmd->llmd_mdt_list);
5058                         list_add_tail(&ltd->ltd_layout_phase_list,
5059                                       &llmd->llmd_mdt_phase2_list);
5060                 }
5061                 break;
5062         case LE_PHASE2_DONE:
5063                 ltd->ltd_layout_done = 1;
5064                 list_del_init(&ltd->ltd_layout_list);
5065                 break;
5066         case LE_PEER_EXIT:
5067                 fail = true;
5068                 ltd->ltd_layout_done = 1;
5069                 list_del_init(&ltd->ltd_layout_list);
5070                 if (!(lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT)) {
5071                         CWARN("%s: the peer %s %x exit layout LFSCK.\n",
5072                               lfsck_lfsck2name(lfsck),
5073                               (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5074                               ltd->ltd_index);
5075                         lo->ll_flags |= LF_INCOMPLETE;
5076                 }
5077                 break;
5078         default:
5079                 break;
5080         }
5081         spin_unlock(&ltds->ltd_lock);
5082
5083         if (fail && lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5084                 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5085
5086                 memset(stop, 0, sizeof(*stop));
5087                 stop->ls_status = lr->lr_status;
5088                 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5089                 lfsck_stop(env, lfsck->li_bottom, stop);
5090         } else if (lfsck_layout_master_to_orphan(llmd)) {
5091                 wake_up_all(&llmd->llmd_thread.t_ctl_waitq);
5092         }
5093
5094         RETURN(0);
5095 }
5096
5097 static int lfsck_layout_slave_in_notify(const struct lu_env *env,
5098                                         struct lfsck_component *com,
5099                                         struct lfsck_request *lr)
5100 {
5101         struct lfsck_instance            *lfsck = com->lc_lfsck;
5102         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
5103         struct lfsck_layout_slave_target *llst;
5104         ENTRY;
5105
5106         if (lr->lr_event == LE_FID_ACCESSED) {
5107                 lfsck_rbtree_update_bitmap(env, com, &lr->lr_fid, true);
5108
5109                 RETURN(0);
5110         }
5111
5112         if (lr->lr_event == LE_CONDITIONAL_DESTROY) {
5113                 int rc;
5114
5115                 rc = lfsck_layout_slave_conditional_destroy(env, com, lr);
5116
5117                 RETURN(rc);
5118         }
5119
5120         if (lr->lr_event != LE_PHASE2_DONE && lr->lr_event != LE_PEER_EXIT)
5121                 RETURN(-EINVAL);
5122
5123         llst = lfsck_layout_llst_find_and_del(llsd, lr->lr_index, true);
5124         if (llst == NULL)
5125                 RETURN(-ENODEV);
5126
5127         lfsck_layout_llst_put(llst);
5128         if (list_empty(&llsd->llsd_master_list))
5129                 wake_up_all(&lfsck->li_thread.t_ctl_waitq);
5130
5131         if (lr->lr_event == LE_PEER_EXIT &&
5132             lfsck->li_bookmark_ram.lb_param & LPF_FAILOUT) {
5133                 struct lfsck_stop *stop = &lfsck_env_info(env)->lti_stop;
5134
5135                 memset(stop, 0, sizeof(*stop));
5136                 stop->ls_status = lr->lr_status;
5137                 stop->ls_flags = lr->lr_param & ~LPF_BROADCAST;
5138                 lfsck_stop(env, lfsck->li_bottom, stop);
5139         }
5140
5141         RETURN(0);
5142 }
5143
5144 static int lfsck_layout_query(const struct lu_env *env,
5145                               struct lfsck_component *com)
5146 {
5147         struct lfsck_layout *lo = com->lc_file_ram;
5148
5149         return lo->ll_status;
5150 }
5151
5152 static int lfsck_layout_master_stop_notify(const struct lu_env *env,
5153                                            struct lfsck_component *com,
5154                                            struct lfsck_tgt_descs *ltds,
5155                                            struct lfsck_tgt_desc *ltd,
5156                                            struct ptlrpc_request_set *set)
5157 {
5158         struct lfsck_thread_info          *info  = lfsck_env_info(env);
5159         struct lfsck_async_interpret_args *laia  = &info->lti_laia;
5160         struct lfsck_request              *lr    = &info->lti_lr;
5161         struct lfsck_instance             *lfsck = com->lc_lfsck;
5162         int                                rc;
5163
5164         spin_lock(&ltds->ltd_lock);
5165         if (list_empty(&ltd->ltd_layout_list)) {
5166                 LASSERT(list_empty(&ltd->ltd_layout_phase_list));
5167                 spin_unlock(&ltds->ltd_lock);
5168
5169                 return 0;
5170         }
5171
5172         list_del_init(&ltd->ltd_layout_phase_list);
5173         list_del_init(&ltd->ltd_layout_list);
5174         spin_unlock(&ltds->ltd_lock);
5175
5176         memset(lr, 0, sizeof(*lr));
5177         lr->lr_index = lfsck_dev_idx(lfsck->li_bottom);
5178         lr->lr_event = LE_PEER_EXIT;
5179         lr->lr_active = LT_LAYOUT;
5180         lr->lr_status = LS_CO_PAUSED;
5181         if (ltds == &lfsck->li_ost_descs)
5182                 lr->lr_flags = LEF_TO_OST;
5183
5184         laia->laia_com = com;
5185         laia->laia_ltds = ltds;
5186         atomic_inc(&ltd->ltd_ref);
5187         laia->laia_ltd = ltd;
5188         laia->laia_lr = lr;
5189         laia->laia_shared = 0;
5190
5191         rc = lfsck_async_request(env, ltd->ltd_exp, lr, set,
5192                                  lfsck_layout_master_async_interpret,
5193                                  laia, LFSCK_NOTIFY);
5194         if (rc != 0) {
5195                 CERROR("%s: Fail to notify %s %x for co-stop: rc = %d\n",
5196                        lfsck_lfsck2name(lfsck),
5197                        (lr->lr_flags & LEF_TO_OST) ? "OST" : "MDT",
5198                        ltd->ltd_index, rc);
5199                 lfsck_tgt_put(ltd);
5200         }
5201
5202         return rc;
5203 }
5204
5205 /* with lfsck::li_lock held */
5206 static int lfsck_layout_slave_join(const struct lu_env *env,
5207                                    struct lfsck_component *com,
5208                                    struct lfsck_start_param *lsp)
5209 {
5210         struct lfsck_instance            *lfsck = com->lc_lfsck;
5211         struct lfsck_layout_slave_data   *llsd  = com->lc_data;
5212         struct lfsck_layout_slave_target *llst;
5213         struct lfsck_start               *start = lsp->lsp_start;
5214         int                               rc    = 0;
5215         ENTRY;
5216
5217         if (!lsp->lsp_index_valid || start == NULL ||
5218             !(start->ls_flags & LPF_ALL_TGT) ||
5219             !(lfsck->li_bookmark_ram.lb_param & LPF_ALL_TGT))
5220                 RETURN(-EALREADY);
5221
5222         spin_unlock(&lfsck->li_lock);
5223         rc = lfsck_layout_llst_add(llsd, lsp->lsp_index);
5224         spin_lock(&lfsck->li_lock);
5225         if (rc == 0 && !thread_is_running(&lfsck->li_thread)) {
5226                 spin_unlock(&lfsck->li_lock);
5227                 llst = lfsck_layout_llst_find_and_del(llsd, lsp->lsp_index,
5228                                                       true);
5229                 if (llst != NULL)
5230                         lfsck_layout_llst_put(llst);
5231                 spin_lock(&lfsck->li_lock);
5232                 rc = -EAGAIN;
5233         }
5234
5235         RETURN(rc);
5236 }
5237
5238 static struct lfsck_operations lfsck_layout_master_ops = {
5239         .lfsck_reset            = lfsck_layout_reset,
5240         .lfsck_fail             = lfsck_layout_fail,
5241         .lfsck_checkpoint       = lfsck_layout_master_checkpoint,
5242         .lfsck_prep             = lfsck_layout_master_prep,
5243         .lfsck_exec_oit         = lfsck_layout_master_exec_oit,
5244         .lfsck_exec_dir         = lfsck_layout_exec_dir,
5245         .lfsck_post             = lfsck_layout_master_post,
5246         .lfsck_interpret        = lfsck_layout_master_async_interpret,
5247         .lfsck_dump             = lfsck_layout_dump,
5248         .lfsck_double_scan      = lfsck_layout_master_double_scan,
5249         .lfsck_data_release     = lfsck_layout_master_data_release,
5250         .lfsck_quit             = lfsck_layout_master_quit,
5251         .lfsck_in_notify        = lfsck_layout_master_in_notify,
5252         .lfsck_query            = lfsck_layout_query,
5253         .lfsck_stop_notify      = lfsck_layout_master_stop_notify,
5254 };
5255
5256 static struct lfsck_operations lfsck_layout_slave_ops = {
5257         .lfsck_reset            = lfsck_layout_reset,
5258         .lfsck_fail             = lfsck_layout_fail,
5259         .lfsck_checkpoint       = lfsck_layout_slave_checkpoint,
5260         .lfsck_prep             = lfsck_layout_slave_prep,
5261         .lfsck_exec_oit         = lfsck_layout_slave_exec_oit,
5262         .lfsck_exec_dir         = lfsck_layout_exec_dir,
5263         .lfsck_post             = lfsck_layout_slave_post,
5264         .lfsck_dump             = lfsck_layout_dump,
5265         .lfsck_double_scan      = lfsck_layout_slave_double_scan,
5266         .lfsck_data_release     = lfsck_layout_slave_data_release,
5267         .lfsck_quit             = lfsck_layout_slave_quit,
5268         .lfsck_in_notify        = lfsck_layout_slave_in_notify,
5269         .lfsck_query            = lfsck_layout_query,
5270         .lfsck_join             = lfsck_layout_slave_join,
5271 };
5272
5273 int lfsck_layout_setup(const struct lu_env *env, struct lfsck_instance *lfsck)
5274 {
5275         struct lfsck_component  *com;
5276         struct lfsck_layout     *lo;
5277         struct dt_object        *root = NULL;
5278         struct dt_object        *obj;
5279         int                      rc;
5280         ENTRY;
5281
5282         OBD_ALLOC_PTR(com);
5283         if (com == NULL)
5284                 RETURN(-ENOMEM);
5285
5286         INIT_LIST_HEAD(&com->lc_link);
5287         INIT_LIST_HEAD(&com->lc_link_dir);
5288         init_rwsem(&com->lc_sem);
5289         atomic_set(&com->lc_ref, 1);
5290         com->lc_lfsck = lfsck;
5291         com->lc_type = LT_LAYOUT;
5292         if (lfsck->li_master) {
5293                 struct lfsck_layout_master_data *llmd;
5294
5295                 com->lc_ops = &lfsck_layout_master_ops;
5296                 OBD_ALLOC_PTR(llmd);
5297                 if (llmd == NULL)
5298                         GOTO(out, rc = -ENOMEM);
5299
5300                 INIT_LIST_HEAD(&llmd->llmd_req_list);
5301                 spin_lock_init(&llmd->llmd_lock);
5302                 INIT_LIST_HEAD(&llmd->llmd_ost_list);
5303                 INIT_LIST_HEAD(&llmd->llmd_ost_phase1_list);
5304                 INIT_LIST_HEAD(&llmd->llmd_ost_phase2_list);
5305                 INIT_LIST_HEAD(&llmd->llmd_mdt_list);
5306                 INIT_LIST_HEAD(&llmd->llmd_mdt_phase1_list);
5307                 INIT_LIST_HEAD(&llmd->llmd_mdt_phase2_list);
5308                 init_waitqueue_head(&llmd->llmd_thread.t_ctl_waitq);
5309                 com->lc_data = llmd;
5310         } else {
5311                 struct lfsck_layout_slave_data *llsd;
5312
5313                 com->lc_ops = &lfsck_layout_slave_ops;
5314                 OBD_ALLOC_PTR(llsd);
5315                 if (llsd == NULL)
5316                         GOTO(out, rc = -ENOMEM);
5317
5318                 INIT_LIST_HEAD(&llsd->llsd_seq_list);
5319                 INIT_LIST_HEAD(&llsd->llsd_master_list);
5320                 spin_lock_init(&llsd->llsd_lock);
5321                 llsd->llsd_rb_root = RB_ROOT;
5322                 rwlock_init(&llsd->llsd_rb_lock);
5323                 com->lc_data = llsd;
5324         }
5325         com->lc_file_size = sizeof(*lo);
5326         OBD_ALLOC(com->lc_file_ram, com->lc_file_size);
5327         if (com->lc_file_ram == NULL)
5328                 GOTO(out, rc = -ENOMEM);
5329
5330         OBD_ALLOC(com->lc_file_disk, com->lc_file_size);
5331         if (com->lc_file_disk == NULL)
5332                 GOTO(out, rc = -ENOMEM);
5333
5334         root = dt_locate(env, lfsck->li_bottom, &lfsck->li_local_root_fid);
5335         if (IS_ERR(root))
5336                 GOTO(out, rc = PTR_ERR(root));
5337
5338         if (unlikely(!dt_try_as_dir(env, root)))
5339                 GOTO(out, rc = -ENOTDIR);
5340
5341         obj = local_file_find_or_create(env, lfsck->li_los, root,
5342                                         lfsck_layout_name,
5343                                         S_IFREG | S_IRUGO | S_IWUSR);
5344         if (IS_ERR(obj))
5345                 GOTO(out, rc = PTR_ERR(obj));
5346
5347         com->lc_obj = obj;
5348         rc = lfsck_layout_load(env, com);
5349         if (rc > 0)
5350                 rc = lfsck_layout_reset(env, com, true);
5351         else if (rc == -ENOENT)
5352                 rc = lfsck_layout_init(env, com);
5353
5354         if (rc != 0)
5355                 GOTO(out, rc);
5356
5357         lo = com->lc_file_ram;
5358         switch (lo->ll_status) {
5359         case LS_INIT:
5360         case LS_COMPLETED:
5361         case LS_FAILED:
5362         case LS_STOPPED:
5363         case LS_PARTIAL:
5364                 spin_lock(&lfsck->li_lock);
5365                 list_add_tail(&com->lc_link, &lfsck->li_list_idle);
5366                 spin_unlock(&lfsck->li_lock);
5367                 break;
5368         default:
5369                 CERROR("%s: unknown lfsck_layout status: rc = %u\n",
5370                        lfsck_lfsck2name(lfsck), lo->ll_status);
5371                 /* fall through */
5372         case LS_SCANNING_PHASE1:
5373         case LS_SCANNING_PHASE2:
5374                 /* No need to store the status to disk right now.
5375                  * If the system crashed before the status stored,
5376                  * it will be loaded back when next time. */
5377                 lo->ll_status = LS_CRASHED;
5378                 lo->ll_flags |= LF_INCOMPLETE;
5379                 /* fall through */
5380         case LS_PAUSED:
5381         case LS_CRASHED:
5382         case LS_CO_FAILED:
5383         case LS_CO_STOPPED:
5384         case LS_CO_PAUSED:
5385                 spin_lock(&lfsck->li_lock);
5386                 list_add_tail(&com->lc_link, &lfsck->li_list_scan);
5387                 spin_unlock(&lfsck->li_lock);
5388                 break;
5389         }
5390
5391         if (lo->ll_flags & LF_CRASHED_LASTID) {
5392                 LASSERT(lfsck->li_out_notify != NULL);
5393
5394                 lfsck->li_out_notify(env, lfsck->li_out_notify_data,
5395                                      LE_LASTID_REBUILDING);
5396         }
5397
5398         GOTO(out, rc = 0);
5399
5400 out:
5401         if (root != NULL && !IS_ERR(root))
5402                 lu_object_put(env, &root->do_lu);
5403
5404         if (rc != 0)
5405                 lfsck_component_cleanup(env, com);
5406
5407         return rc;
5408 }
5409
5410 struct lfsck_orphan_it {
5411         struct lfsck_component           *loi_com;
5412         struct lfsck_rbtree_node         *loi_lrn;
5413         struct lfsck_layout_slave_target *loi_llst;
5414         struct lu_fid                     loi_key;
5415         struct lu_orphan_rec              loi_rec;
5416         __u64                             loi_hash;
5417         unsigned int                      loi_over:1;
5418 };
5419
5420 static int lfsck_fid_match_idx(const struct lu_env *env,
5421                                struct lfsck_instance *lfsck,
5422                                const struct lu_fid *fid, int idx)
5423 {
5424         struct seq_server_site  *ss;
5425         struct lu_server_fld    *sf;
5426         struct lu_seq_range      range  = { 0 };
5427         int                      rc;
5428
5429         /* All abnormal cases will be returned to MDT0. */
5430         if (!fid_is_norm(fid)) {
5431                 if (idx == 0)
5432                         return 1;
5433
5434                 return 0;
5435         }
5436
5437         ss = lu_site2seq(lfsck->li_bottom->dd_lu_dev.ld_site);
5438         if (unlikely(ss == NULL))
5439                 return -ENOTCONN;
5440
5441         sf = ss->ss_server_fld;
5442         LASSERT(sf != NULL);
5443
5444         fld_range_set_any(&range);
5445         rc = fld_server_lookup(env, sf, fid_seq(fid), &range);
5446         if (rc != 0)
5447                 return rc;
5448
5449         if (!fld_range_is_mdt(&range))
5450                 return -EINVAL;
5451
5452         if (range.lsr_index == idx)
5453                 return 1;
5454
5455         return 0;
5456 }
5457
5458 static void lfsck_layout_destroy_orphan(const struct lu_env *env,
5459                                         struct dt_device *dev,
5460                                         struct dt_object *obj)
5461 {
5462         struct thandle *handle;
5463         int             rc;
5464         ENTRY;
5465
5466         handle = dt_trans_create(env, dev);
5467         if (IS_ERR(handle))
5468                 RETURN_EXIT;
5469
5470         rc = dt_declare_ref_del(env, obj, handle);
5471         if (rc != 0)
5472                 GOTO(stop, rc);
5473
5474         rc = dt_declare_destroy(env, obj, handle);
5475         if (rc != 0)
5476                 GOTO(stop, rc);
5477
5478         rc = dt_trans_start_local(env, dev, handle);
5479         if (rc != 0)
5480                 GOTO(stop, rc);
5481
5482         dt_write_lock(env, obj, 0);
5483         rc = dt_ref_del(env, obj, handle);
5484         if (rc == 0)
5485                 rc = dt_destroy(env, obj, handle);
5486         dt_write_unlock(env, obj);
5487
5488         GOTO(stop, rc);
5489
5490 stop:
5491         dt_trans_stop(env, dev, handle);
5492
5493         RETURN_EXIT;
5494 }
5495
5496 static int lfsck_orphan_index_lookup(const struct lu_env *env,
5497                                      struct dt_object *dt,
5498                                      struct dt_rec *rec,
5499                                      const struct dt_key *key,
5500                                      struct lustre_capa *capa)
5501 {
5502         return -EOPNOTSUPP;
5503 }
5504
5505 static int lfsck_orphan_index_declare_insert(const struct lu_env *env,
5506                                              struct dt_object *dt,
5507                                              const struct dt_rec *rec,
5508                                              const struct dt_key *key,
5509                                              struct thandle *handle)
5510 {
5511         return -EOPNOTSUPP;
5512 }
5513
5514 static int lfsck_orphan_index_insert(const struct lu_env *env,
5515                                      struct dt_object *dt,
5516                                      const struct dt_rec *rec,
5517                                      const struct dt_key *key,
5518                                      struct thandle *handle,
5519                                      struct lustre_capa *capa,
5520                                      int ignore_quota)
5521 {
5522         return -EOPNOTSUPP;
5523 }
5524
5525 static int lfsck_orphan_index_declare_delete(const struct lu_env *env,
5526                                              struct dt_object *dt,
5527                                              const struct dt_key *key,
5528                                              struct thandle *handle)
5529 {
5530         return -EOPNOTSUPP;
5531 }
5532
5533 static int lfsck_orphan_index_delete(const struct lu_env *env,
5534                                      struct dt_object *dt,
5535                                      const struct dt_key *key,
5536                                      struct thandle *handle,
5537                                      struct lustre_capa *capa)
5538 {
5539         return -EOPNOTSUPP;
5540 }
5541
5542 static struct dt_it *lfsck_orphan_it_init(const struct lu_env *env,
5543                                           struct dt_object *dt,
5544                                           __u32 attr,
5545                                           struct lustre_capa *capa)
5546 {
5547         struct dt_device                *dev    = lu2dt_dev(dt->do_lu.lo_dev);
5548         struct lfsck_instance           *lfsck;
5549         struct lfsck_component          *com    = NULL;
5550         struct lfsck_layout_slave_data  *llsd;
5551         struct lfsck_orphan_it          *it     = NULL;
5552         int                              rc     = 0;
5553         ENTRY;
5554
5555         lfsck = lfsck_instance_find(dev, true, false);
5556         if (unlikely(lfsck == NULL))
5557                 RETURN(ERR_PTR(-ENODEV));
5558
5559         com = lfsck_component_find(lfsck, LT_LAYOUT);
5560         if (unlikely(com == NULL))
5561                 GOTO(out, rc = -ENOENT);
5562
5563         llsd = com->lc_data;
5564         if (!llsd->llsd_rbtree_valid)
5565                 GOTO(out, rc = -ESRCH);
5566
5567         OBD_ALLOC_PTR(it);
5568         if (it == NULL)
5569                 GOTO(out, rc = -ENOMEM);
5570
5571         it->loi_llst = lfsck_layout_llst_find_and_del(llsd, attr, false);
5572         if (it->loi_llst == NULL)
5573                 GOTO(out, rc = -ENODEV);
5574
5575         if (dev->dd_record_fid_accessed) {
5576                 /* The first iteration against the rbtree, scan the whole rbtree
5577                  * to remove the nodes which do NOT need to be handled. */
5578                 write_lock(&llsd->llsd_rb_lock);
5579                 if (dev->dd_record_fid_accessed) {
5580                         struct rb_node                  *node;
5581                         struct rb_node                  *next;
5582                         struct lfsck_rbtree_node        *lrn;
5583
5584                         /* No need to record the fid accessing anymore. */
5585                         dev->dd_record_fid_accessed = 0;
5586
5587                         node = rb_first(&llsd->llsd_rb_root);
5588                         while (node != NULL) {
5589                                 next = rb_next(node);
5590                                 lrn = rb_entry(node, struct lfsck_rbtree_node,
5591                                                lrn_node);
5592                                 if (atomic_read(&lrn->lrn_known_count) <=
5593                                     atomic_read(&lrn->lrn_accessed_count)) {
5594                                         rb_erase(node, &llsd->llsd_rb_root);
5595                                         lfsck_rbtree_free(lrn);
5596                                 }
5597                                 node = next;
5598                         }
5599                 }
5600                 write_unlock(&llsd->llsd_rb_lock);
5601         }
5602
5603         /* read lock the rbtree when init, and unlock when fini */
5604         read_lock(&llsd->llsd_rb_lock);
5605         it->loi_com = com;
5606         com = NULL;
5607
5608         GOTO(out, rc = 0);
5609
5610 out:
5611         if (com != NULL)
5612                 lfsck_component_put(env, com);
5613         lfsck_instance_put(env, lfsck);
5614         if (rc != 0) {
5615                 if (it != NULL)
5616                         OBD_FREE_PTR(it);
5617
5618                 it = (struct lfsck_orphan_it *)ERR_PTR(rc);
5619         }
5620
5621         return (struct dt_it *)it;
5622 }
5623
5624 static void lfsck_orphan_it_fini(const struct lu_env *env,
5625                                  struct dt_it *di)
5626 {
5627         struct lfsck_orphan_it           *it    = (struct lfsck_orphan_it *)di;
5628         struct lfsck_component           *com   = it->loi_com;
5629         struct lfsck_layout_slave_data   *llsd;
5630         struct lfsck_layout_slave_target *llst;
5631
5632         if (com != NULL) {
5633                 llsd = com->lc_data;
5634                 read_unlock(&llsd->llsd_rb_lock);
5635                 llst = it->loi_llst;
5636                 LASSERT(llst != NULL);
5637
5638                 /* Save the key and hash for iterate next. */
5639                 llst->llst_fid = it->loi_key;
5640                 llst->llst_hash = it->loi_hash;
5641                 lfsck_layout_llst_put(llst);
5642                 lfsck_component_put(env, com);
5643         }
5644         OBD_FREE_PTR(it);
5645 }
5646
5647 /**
5648  * \retval       +1: the iteration finished
5649  * \retval        0: on success, not finished
5650  * \retval      -ve: on error
5651  */
5652 static int lfsck_orphan_it_next(const struct lu_env *env,
5653                                 struct dt_it *di)
5654 {
5655         struct lfsck_thread_info        *info   = lfsck_env_info(env);
5656         struct filter_fid_old           *pfid   = &info->lti_old_pfid;
5657         struct lu_attr                  *la     = &info->lti_la;
5658         struct lfsck_orphan_it          *it     = (struct lfsck_orphan_it *)di;
5659         struct lu_fid                   *key    = &it->loi_key;
5660         struct lu_orphan_rec            *rec    = &it->loi_rec;
5661         struct lfsck_component          *com    = it->loi_com;
5662         struct lfsck_instance           *lfsck  = com->lc_lfsck;
5663         struct lfsck_layout_slave_data  *llsd   = com->lc_data;
5664         struct dt_object                *obj;
5665         struct lfsck_rbtree_node        *lrn;
5666         int                              pos;
5667         int                              rc;
5668         __u32                            save;
5669         __u32                            idx    = it->loi_llst->llst_index;
5670         bool                             exact  = false;
5671         ENTRY;
5672
5673         if (it->loi_over)
5674                 RETURN(1);
5675
5676 again0:
5677         lrn = it->loi_lrn;
5678         if (lrn == NULL) {
5679                 lrn = lfsck_rbtree_search(llsd, key, &exact);
5680                 if (lrn == NULL) {
5681                         it->loi_over = 1;
5682                         RETURN(1);
5683                 }
5684
5685                 it->loi_lrn = lrn;
5686                 if (!exact) {
5687                         key->f_seq = lrn->lrn_seq;
5688                         key->f_oid = lrn->lrn_first_oid;
5689                         key->f_ver = 0;
5690                 }
5691         } else {
5692                 key->f_oid++;
5693                 if (unlikely(key->f_oid == 0)) {
5694                         key->f_seq++;
5695                         it->loi_lrn = NULL;
5696                         goto again0;
5697                 }
5698
5699                 if (key->f_oid >=
5700                     lrn->lrn_first_oid + LFSCK_RBTREE_BITMAP_WIDTH) {
5701                         it->loi_lrn = NULL;
5702                         goto again0;
5703                 }
5704         }
5705
5706         if (unlikely(atomic_read(&lrn->lrn_known_count) <=
5707                      atomic_read(&lrn->lrn_accessed_count))) {
5708                 struct rb_node *next = rb_next(&lrn->lrn_node);
5709
5710                 while (next != NULL) {
5711                         lrn = rb_entry(next, struct lfsck_rbtree_node,
5712                                        lrn_node);
5713                         if (atomic_read(&lrn->lrn_known_count) >
5714                             atomic_read(&lrn->lrn_accessed_count))
5715                                 break;
5716                         next = rb_next(next);
5717                 }
5718
5719                 if (next == NULL) {
5720                         it->loi_over = 1;
5721                         RETURN(1);
5722                 }
5723
5724                 it->loi_lrn = lrn;
5725                 key->f_seq = lrn->lrn_seq;
5726                 key->f_oid = lrn->lrn_first_oid;
5727                 key->f_ver = 0;
5728         }
5729
5730         pos = key->f_oid - lrn->lrn_first_oid;
5731
5732 again1:
5733         pos = find_next_bit(lrn->lrn_known_bitmap,
5734                             LFSCK_RBTREE_BITMAP_WIDTH, pos);
5735         if (pos >= LFSCK_RBTREE_BITMAP_WIDTH) {
5736                 key->f_oid = lrn->lrn_first_oid + pos;
5737                 if (unlikely(key->f_oid < lrn->lrn_first_oid)) {
5738                         key->f_seq++;
5739                         key->f_oid = 0;
5740                 }
5741                 it->loi_lrn = NULL;
5742                 goto again0;
5743         }
5744
5745         if (test_bit(pos, lrn->lrn_accessed_bitmap)) {
5746                 pos++;
5747                 goto again1;
5748         }
5749
5750         key->f_oid = lrn->lrn_first_oid + pos;
5751         obj = lfsck_object_find(env, lfsck, key);
5752         if (IS_ERR(obj)) {
5753                 rc = PTR_ERR(obj);
5754                 if (rc == -ENOENT) {
5755                         pos++;
5756                         goto again1;
5757                 }
5758                 RETURN(rc);
5759         }
5760
5761         dt_read_lock(env, obj, 0);
5762         if (!dt_object_exists(obj)) {
5763                 dt_read_unlock(env, obj);
5764                 lfsck_object_put(env, obj);
5765                 pos++;
5766                 goto again1;
5767         }
5768
5769         rc = dt_attr_get(env, obj, la, BYPASS_CAPA);
5770         if (rc != 0)
5771                 GOTO(out, rc);
5772
5773         rc = dt_xattr_get(env, obj, lfsck_buf_get(env, pfid, sizeof(*pfid)),
5774                           XATTR_NAME_FID, BYPASS_CAPA);
5775         if (rc == -ENODATA) {
5776                 /* For the pre-created OST-object, update the bitmap to avoid
5777                  * others LFSCK (second phase) iteration to touch it again. */
5778                 if (la->la_ctime == 0) {
5779                         if (!test_and_set_bit(pos, lrn->lrn_accessed_bitmap))
5780                                 atomic_inc(&lrn->lrn_accessed_count);
5781
5782                         /* For the race between repairing dangling referenced
5783                          * MDT-object and unlink the file, it may left orphan
5784                          * OST-object there. Destroy it now! */
5785                         if (unlikely(!(la->la_mode & S_ISUID))) {
5786                                 dt_read_unlock(env, obj);
5787                                 lfsck_layout_destroy_orphan(env,
5788                                                             lfsck->li_bottom,
5789                                                             obj);
5790                                 lfsck_object_put(env, obj);
5791                                 pos++;
5792                                 goto again1;
5793                         }
5794                 } else if (idx == 0) {
5795                         /* If the orphan OST-object has no parent information,
5796                          * regard it as referenced by the MDT-object on MDT0. */
5797                         fid_zero(&rec->lor_fid);
5798                         rec->lor_uid = la->la_uid;
5799                         rec->lor_gid = la->la_gid;
5800                         GOTO(out, rc = 0);
5801                 }
5802
5803                 dt_read_unlock(env, obj);
5804                 lfsck_object_put(env, obj);
5805                 pos++;
5806                 goto again1;
5807         }
5808
5809         if (rc < 0)
5810                 GOTO(out, rc);
5811
5812         if (rc != sizeof(struct filter_fid) &&
5813             rc != sizeof(struct filter_fid_old))
5814                 GOTO(out, rc = -EINVAL);
5815
5816         fid_le_to_cpu(&rec->lor_fid, &pfid->ff_parent);
5817         /* In fact, the ff_parent::f_ver is not the real parent FID::f_ver,
5818          * instead, it is the OST-object index in its parent MDT-object
5819          * layout EA. */
5820         save = rec->lor_fid.f_ver;
5821         rec->lor_fid.f_ver = 0;
5822         rc = lfsck_fid_match_idx(env, lfsck, &rec->lor_fid, idx);
5823         /* If the orphan OST-object does not claim the MDT, then next.
5824          *
5825          * If we do not know whether it matches or not, then return it
5826          * to the MDT for further check. */
5827         if (rc == 0) {
5828                 dt_read_unlock(env, obj);
5829                 lfsck_object_put(env, obj);
5830                 pos++;
5831                 goto again1;
5832         }
5833
5834         rec->lor_fid.f_ver = save;
5835         rec->lor_uid = la->la_uid;
5836         rec->lor_gid = la->la_gid;
5837
5838         CDEBUG(D_LFSCK, "%s: return orphan "DFID", PFID "DFID", owner %u:%u\n",
5839                lfsck_lfsck2name(com->lc_lfsck), PFID(key), PFID(&rec->lor_fid),
5840                rec->lor_uid, rec->lor_gid);
5841
5842         GOTO(out, rc = 0);
5843
5844 out:
5845         dt_read_unlock(env, obj);
5846         lfsck_object_put(env, obj);
5847         if (rc == 0)
5848                 it->loi_hash++;
5849
5850         return rc;
5851 }
5852
5853 /**
5854  * \retval       +1: locate to the exactly position
5855  * \retval        0: cannot locate to the exactly position,
5856  *                   call next() to move to a valid position.
5857  * \retval      -ve: on error
5858  */
5859 static int lfsck_orphan_it_get(const struct lu_env *env,
5860                                struct dt_it *di,
5861                                const struct dt_key *key)
5862 {
5863         struct lfsck_orphan_it  *it   = (struct lfsck_orphan_it *)di;
5864         int                      rc;
5865
5866         it->loi_key = *(struct lu_fid *)key;
5867         rc = lfsck_orphan_it_next(env, di);
5868         if (rc == 1)
5869                 return 0;
5870
5871         if (rc == 0)
5872                 return 1;
5873
5874         return rc;
5875 }
5876
5877 static void lfsck_orphan_it_put(const struct lu_env *env,
5878                                 struct dt_it *di)
5879 {
5880 }
5881
5882 static struct dt_key *lfsck_orphan_it_key(const struct lu_env *env,
5883                                           const struct dt_it *di)
5884 {
5885         struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5886
5887         return (struct dt_key *)&it->loi_key;
5888 }
5889
5890 static int lfsck_orphan_it_key_size(const struct lu_env *env,
5891                                     const struct dt_it *di)
5892 {
5893         return sizeof(struct lu_fid);
5894 }
5895
5896 static int lfsck_orphan_it_rec(const struct lu_env *env,
5897                                const struct dt_it *di,
5898                                struct dt_rec *rec,
5899                                __u32 attr)
5900 {
5901         struct lfsck_orphan_it *it = (struct lfsck_orphan_it *)di;
5902
5903         *(struct lu_orphan_rec *)rec = it->loi_rec;
5904
5905         return 0;
5906 }
5907
5908 static __u64 lfsck_orphan_it_store(const struct lu_env *env,
5909                                    const struct dt_it *di)
5910 {
5911         struct lfsck_orphan_it  *it   = (struct lfsck_orphan_it *)di;
5912
5913         return it->loi_hash;
5914 }
5915
5916 /**
5917  * \retval       +1: locate to the exactly position
5918  * \retval        0: cannot locate to the exactly position,
5919  *                   call next() to move to a valid position.
5920  * \retval      -ve: on error
5921  */
5922 static int lfsck_orphan_it_load(const struct lu_env *env,
5923                                 const struct dt_it *di,
5924                                 __u64 hash)
5925 {
5926         struct lfsck_orphan_it           *it   = (struct lfsck_orphan_it *)di;
5927         struct lfsck_layout_slave_target *llst = it->loi_llst;
5928         int                               rc;
5929
5930         LASSERT(llst != NULL);
5931
5932         if (hash != llst->llst_hash) {
5933                 CWARN("%s: the given hash "LPU64" for orphan iteration does "
5934                       "not match the one when fini "LPU64", to be reset.\n",
5935                       lfsck_lfsck2name(it->loi_com->lc_lfsck), hash,
5936                       llst->llst_hash);
5937                 fid_zero(&llst->llst_fid);
5938                 llst->llst_hash = 0;
5939         }
5940
5941         it->loi_key = llst->llst_fid;
5942         it->loi_hash = llst->llst_hash;
5943         rc = lfsck_orphan_it_next(env, (struct dt_it *)di);
5944         if (rc == 1)
5945                 return 0;
5946
5947         if (rc == 0)
5948                 return 1;
5949
5950         return rc;
5951 }
5952
5953 static int lfsck_orphan_it_key_rec(const struct lu_env *env,
5954                                    const struct dt_it *di,
5955                                    void *key_rec)
5956 {
5957         return 0;
5958 }
5959
5960 const struct dt_index_operations lfsck_orphan_index_ops = {
5961         .dio_lookup             = lfsck_orphan_index_lookup,
5962         .dio_declare_insert     = lfsck_orphan_index_declare_insert,
5963         .dio_insert             = lfsck_orphan_index_insert,
5964         .dio_declare_delete     = lfsck_orphan_index_declare_delete,
5965         .dio_delete             = lfsck_orphan_index_delete,
5966         .dio_it = {
5967                 .init           = lfsck_orphan_it_init,
5968                 .fini           = lfsck_orphan_it_fini,
5969                 .get            = lfsck_orphan_it_get,
5970                 .put            = lfsck_orphan_it_put,
5971                 .next           = lfsck_orphan_it_next,
5972                 .key            = lfsck_orphan_it_key,
5973                 .key_size       = lfsck_orphan_it_key_size,
5974                 .rec            = lfsck_orphan_it_rec,
5975                 .store          = lfsck_orphan_it_store,
5976                 .load           = lfsck_orphan_it_load,
5977                 .key_rec        = lfsck_orphan_it_key_rec,
5978         }
5979 };