lustre/lfsck/lfsck_layout.c

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