lustre/lfsck/lfsck_layout.c

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