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