LU-12127 lfsck: skip orphan processing for namespace LFSCK

[fs/lustre-release.git] / lustre / lfsck / lfsck_engine.c

diff --git a/lustre/lfsck/lfsck_engine.c b/lustre/lfsck/lfsck_engine.c
index f5b5e70..a1763a6 100644 (file)
--- a/lustre/lfsck/lfsck_engine.c
+++ b/lustre/lfsck/lfsck_engine.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2013, 2015, Intel Corporation.
+ * Copyright (c) 2013, 2016, Intel Corporation.
  */
 /*
  * lustre/lfsck/lfsck_engine.c
@@ -136,6 +136,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
        struct lfsck_thread_info *info    = lfsck_env_info(env);
        struct lu_fid            *fid     = &info->lti_fid;
        struct lu_seq_range      *range   = &info->lti_range;
+       struct lu_attr           *la      = &info->lti_la;
        struct seq_server_site   *ss      = lfsck_dev_site(lfsck);
        __u32                     idx     = lfsck_dev_idx(lfsck);
        int                       depth   = 0;
@@ -144,9 +145,21 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
        if (list_empty(&lfsck->li_list_dir) || !S_ISDIR(lfsck_object_type(obj)))
                return 0;
 
+       *fid = *lfsck_dto2fid(obj);
+       rc = dt_attr_get(env, obj, la);
+       if (unlikely(rc || (la->la_valid & LA_FLAGS &&
+                           la->la_flags & LUSTRE_ORPHAN_FL))) {
+               /* Orphan directory is empty, does not need scan. */
+               CDEBUG(D_INFO,
+                      "%s: skip orphan dir "DFID", %llx/%x: rc = %d\n",
+                      lfsck_lfsck2name(lfsck), PFID(fid),
+                      la->la_valid, la->la_flags, rc);
+
+               return rc;
+       }
+
        LASSERT(ss != NULL);
 
-       *fid = *lfsck_dto2fid(obj);
        while (1) {
                /* Global /ROOT is visible. */
                if (unlikely(lu_fid_eq(fid, &lfsck->li_global_root_fid)))
@@ -177,6 +190,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
                if (fid_is_norm(fid))
                        return 1;
 
+               /* Only true after "obj = NULL" set below */
                if (obj == NULL) {
                        obj = lfsck_object_find_bottom(env, lfsck, fid);
                        if (IS_ERR(obj))
@@ -187,7 +201,7 @@ static int lfsck_needs_scan_dir(const struct lu_env *env,
                                GOTO(out, rc = 0);
                }
 
-               dt_read_lock(env, obj, MOR_TGT_CHILD);
+               dt_read_lock(env, obj, DT_TGT_CHILD);
                if (unlikely(lfsck_is_dead_obj(obj))) {
                        dt_read_unlock(env, obj);
 
@@ -234,7 +248,7 @@ static int lfsck_load_stripe_lmv(const struct lu_env *env,
        LASSERT(lfsck->li_obj_dir == NULL);
        LASSERT(lfsck->li_lmv == NULL);
 
-       rc = lfsck_read_stripe_lmv(env, obj, lmv);
+       rc = lfsck_read_stripe_lmv(env, lfsck, obj, lmv);
        if (rc == -ENODATA) {
                lfsck->li_obj_dir = lfsck_object_get(obj);
 
@@ -362,7 +376,9 @@ int lfsck_open_dir(const struct lu_env *env,
                GOTO(out, rc = PTR_ERR(di));
 
        rc = iops->load(env, di, cookie);
-       if (rc == 0 || (rc > 0 && cookie > 0))
+       if (rc == -ENODATA)
+               rc = 1;
+       else if (rc == 0 || (rc > 0 && cookie > 0))
                rc = iops->next(env, di);
        else if (rc > 0)
                rc = 0;
@@ -393,8 +409,7 @@ static int lfsck_checkpoint(const struct lu_env *env,
        int                     rc  = 0;
        int                     rc1 = 0;
 
-       if (likely(cfs_time_beforeq(cfs_time_current(),
-                                   lfsck->li_time_next_checkpoint)))
+       if (likely(ktime_get_seconds() <= lfsck->li_time_next_checkpoint))
                return 0;
 
        lfsck_pos_fill(env, lfsck, &lfsck->li_pos_checkpoint, false);
@@ -404,9 +419,9 @@ static int lfsck_checkpoint(const struct lu_env *env,
                        rc1 = rc;
        }
 
-       lfsck->li_time_last_checkpoint = cfs_time_current();
+       lfsck->li_time_last_checkpoint = ktime_get_seconds();
        lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
-                               cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
+                                        LFSCK_CHECKPOINT_INTERVAL;
        return rc1 != 0 ? rc1 : rc;
 }
 
@@ -441,7 +456,7 @@ static int lfsck_prep(const struct lu_env *env, struct lfsck_instance *lfsck,
        /* Init otable-based iterator. */
        if (pos == NULL) {
                rc = iops->load(env, lfsck->li_di_oit, 0);
-               if (rc > 0) {
+               if (rc > 0 || unlikely(rc == -ENODATA)) {
                        lfsck->li_oit_over = 1;
                        rc = 0;
                }
@@ -450,10 +465,10 @@ static int lfsck_prep(const struct lu_env *env, struct lfsck_instance *lfsck,
        }
 
        rc = iops->load(env, lfsck->li_di_oit, pos->lp_oit_cookie);
-       if (rc < 0)
-               GOTO(out, rc);
-       else if (rc > 0)
+       if (rc > 0 || unlikely(rc == -ENODATA))
                lfsck->li_oit_over = 1;
+       else if (rc < 0)
+               GOTO(out, rc);
 
        if (!lfsck->li_master || fid_is_zero(&pos->lp_dir_parent))
                GOTO(out, rc = 0);
@@ -507,9 +522,9 @@ out:
                        break;
        }
 
-       lfsck->li_time_last_checkpoint = cfs_time_current();
+       lfsck->li_time_last_checkpoint = ktime_get_seconds();
        lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
-                               cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
+                                        LFSCK_CHECKPOINT_INTERVAL;
        return rc;
 }
 
@@ -606,9 +621,9 @@ static int lfsck_post(const struct lu_env *env, struct lfsck_instance *lfsck,
                               (__u32)com->lc_type, rc);
        }
 
-       lfsck->li_time_last_checkpoint = cfs_time_current();
+       lfsck->li_time_last_checkpoint = ktime_get_seconds();
        lfsck->li_time_next_checkpoint = lfsck->li_time_last_checkpoint +
-                               cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
+                                        LFSCK_CHECKPOINT_INTERVAL;
 
        /* Ignore some component post failure to make other can go ahead. */
        return result;
@@ -1015,6 +1030,10 @@ int lfsck_master_engine(void *args)
        int                       rc;
        ENTRY;
 
+       spin_lock(&lfsck->li_lock);
+       lfsck->li_task = current;
+       spin_unlock(&lfsck->li_lock);
+
        /* There will be some objects verification during the LFSCK start,
         * such as the subsequent lfsck_verify_lpf(). Trigger low layer OI
         * OI scrub before that to handle the potential inconsistence. */
@@ -1058,6 +1077,11 @@ int lfsck_master_engine(void *args)
               current_pid());
 
        spin_lock(&lfsck->li_lock);
+       if (unlikely(!thread_is_starting(thread))) {
+               spin_unlock(&lfsck->li_lock);
+               GOTO(fini_oit, rc = 0);
+       }
+
        thread_set_flags(thread, SVC_RUNNING);
        spin_unlock(&lfsck->li_lock);
        wake_up_all(&thread->t_ctl_waitq);
@@ -1106,6 +1130,7 @@ fini_oit:
 fini_args:
        spin_lock(&lfsck->li_lock);
        thread_set_flags(thread, SVC_STOPPED);
+       lfsck->li_task = NULL;
        spin_unlock(&lfsck->li_lock);
        wake_up_all(&thread->t_ctl_waitq);
        lfsck_thread_args_fini(lta);
@@ -1224,7 +1249,7 @@ again:
        }
        spin_unlock(&ltds->ltd_lock);
 
-       rc = ptlrpc_set_wait(set);
+       rc = ptlrpc_set_wait(env, set);
        if (rc < 0) {
                ptlrpc_set_destroy(set);
                RETURN(rc);
@@ -1319,7 +1344,7 @@ static int lfsck_assistant_notify_others(const struct lu_env *env,
                up_read(&ltds->ltd_rw_sem);
 
                /* Sync up */
-               rc = ptlrpc_set_wait(set);
+               rc = ptlrpc_set_wait(env, set);
                if (rc < 0) {
                        ptlrpc_set_destroy(set);
                        RETURN(rc);
@@ -1451,7 +1476,7 @@ again:
                }
                spin_unlock(&ltds->ltd_lock);
 
-               rc = ptlrpc_set_wait(set);
+               rc = ptlrpc_set_wait(env, set);
                if (rc < 0) {
                        ptlrpc_set_destroy(set);
                        RETURN(rc);
@@ -1525,7 +1550,7 @@ again:
                break;
        }
 
-       rc1 = ptlrpc_set_wait(set);
+       rc1 = ptlrpc_set_wait(env, set);
        ptlrpc_set_destroy(set);
 
        RETURN(rc != 0 ? rc : rc1);
@@ -1595,7 +1620,7 @@ int lfsck_assistant_engine(void *args)
                while (!list_empty(&lad->lad_req_list)) {
                        bool wakeup = false;
 
-                       if (unlikely(lad->lad_exit ||
+                       if (unlikely(test_bit(LAD_EXIT, &lad->lad_flags) ||
                                     !thread_is_running(mthread)))
                                GOTO(cleanup, rc = lad->lad_post_result);
 
@@ -1627,25 +1652,25 @@ int lfsck_assistant_engine(void *args)
 
                l_wait_event(athread->t_ctl_waitq,
                             !lfsck_assistant_req_empty(lad) ||
-                            lad->lad_exit ||
-                            lad->lad_to_post ||
-                            lad->lad_to_double_scan,
+                            test_bit(LAD_EXIT, &lad->lad_flags) ||
+                            test_bit(LAD_TO_POST, &lad->lad_flags) ||
+                            test_bit(LAD_TO_DOUBLE_SCAN, &lad->lad_flags),
                             &lwi);
 
-               if (unlikely(lad->lad_exit))
+               if (unlikely(test_bit(LAD_EXIT, &lad->lad_flags)))
                        GOTO(cleanup, rc = lad->lad_post_result);
 
                if (!list_empty(&lad->lad_req_list))
                        continue;
 
-               if (lad->lad_to_post) {
+               if (test_bit(LAD_TO_POST, &lad->lad_flags)) {
                        CDEBUG(D_LFSCK, "%s: %s LFSCK assistant thread post\n",
                               lfsck_lfsck2name(lfsck), lad->lad_name);
 
-                       if (unlikely(lad->lad_exit))
+                       if (unlikely(test_bit(LAD_EXIT, &lad->lad_flags)))
                                GOTO(cleanup, rc = lad->lad_post_result);
 
-                       lad->lad_to_post = 0;
+                       clear_bit(LAD_TO_POST, &lad->lad_flags);
                        LASSERT(lad->lad_post_result > 0);
 
                        /* Wakeup the master engine to go ahead. */
@@ -1662,18 +1687,18 @@ int lfsck_assistant_engine(void *args)
                               lad->lad_name, rc);
                }
 
-               if (lad->lad_to_double_scan) {
-                       lad->lad_to_double_scan = 0;
+               if (test_bit(LAD_TO_DOUBLE_SCAN, &lad->lad_flags)) {
+                       clear_bit(LAD_TO_DOUBLE_SCAN, &lad->lad_flags);
                        atomic_inc(&lfsck->li_double_scan_count);
-                       lad->lad_in_double_scan = 1;
+                       set_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags);
                        wake_up_all(&mthread->t_ctl_waitq);
 
                        com->lc_new_checked = 0;
                        com->lc_new_scanned = 0;
-                       com->lc_time_last_checkpoint = cfs_time_current();
+                       com->lc_time_last_checkpoint = ktime_get_seconds();
                        com->lc_time_next_checkpoint =
                                com->lc_time_last_checkpoint +
-                               cfs_time_seconds(LFSCK_CHECKPOINT_INTERVAL);
+                               LFSCK_CHECKPOINT_INTERVAL;
 
                        CDEBUG(D_LFSCK, "%s: LFSCK assistant sync before "
                               "the second-stage scaning\n",
@@ -1689,7 +1714,7 @@ int lfsck_assistant_engine(void *args)
                        if (OBD_FAIL_CHECK(OBD_FAIL_LFSCK_NO_DOUBLESCAN))
                                GOTO(cleanup, rc = 0);
 
-                       while (lad->lad_in_double_scan) {
+                       while (test_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags)) {
                                rc = lfsck_assistant_query_others(env, com);
                                if (lfsck_phase2_next_ready(lad))
                                        goto p2_next;
@@ -1704,12 +1729,13 @@ int lfsck_assistant_engine(void *args)
                                                           NULL, NULL);
                                rc = l_wait_event(athread->t_ctl_waitq,
                                        lfsck_phase2_next_ready(lad) ||
-                                       lad->lad_exit ||
+                                       test_bit(LAD_EXIT, &lad->lad_flags) ||
                                        !thread_is_running(mthread),
                                        &lwi);
 
-                               if (unlikely(lad->lad_exit ||
-                                            !thread_is_running(mthread)))
+                               if (unlikely(
+                                       test_bit(LAD_EXIT, &lad->lad_flags) ||
+                                       !thread_is_running(mthread)))
                                        GOTO(cleanup, rc = 0);
 
                                if (rc == -ETIMEDOUT)
@@ -1723,8 +1749,9 @@ p2_next:
                                if (rc != 0)
                                        GOTO(cleanup, rc);
 
-                               if (unlikely(lad->lad_exit ||
-                                            !thread_is_running(mthread)))
+                               if (unlikely(
+                                       test_bit(LAD_EXIT, &lad->lad_flags) ||
+                                       !thread_is_running(mthread)))
                                        GOTO(cleanup, rc = 0);
                        }
                }
@@ -1736,7 +1763,7 @@ cleanup:
        if (rc < 0)
                lad->lad_assistant_status = rc;
 
-       if (lad->lad_exit && lad->lad_post_result <= 0)
+       if (test_bit(LAD_EXIT, &lad->lad_flags) && lad->lad_post_result <= 0)
                lao->la_fill_pos(env, com, &lfsck->li_pos_checkpoint);
 
        thread_set_flags(athread, SVC_STOPPING);
@@ -1752,9 +1779,6 @@ cleanup:
        }
        spin_unlock(&lad->lad_lock);
 
-       LASSERTF(lad->lad_prefetched == 0, "unmatched prefeteched objs %d\n",
-                lad->lad_prefetched);
-
        memset(lr, 0, sizeof(*lr));
        if (rc > 0) {
                lr->lr_event = LE_PHASE2_DONE;
@@ -1813,8 +1837,8 @@ cleanup:
        /* Under force exit case, some requests may be just freed without
         * verification, those objects should be re-handled when next run.
         * So not update the on-disk trace file under such case. */
-       if (lad->lad_in_double_scan) {
-               if (!lad->lad_exit)
+       if (test_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags)) {
+               if (!test_bit(LAD_EXIT, &lad->lad_flags))
                        rc1 = lao->la_double_scan_result(env, com, rc);
 
                CDEBUG(D_LFSCK, "%s: LFSCK assistant phase2 scan "
@@ -1823,13 +1847,12 @@ cleanup:
        }
 
 fini:
-       if (lad->lad_in_double_scan)
+       if (test_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags))
                atomic_dec(&lfsck->li_double_scan_count);
 
        spin_lock(&lad->lad_lock);
        lad->lad_assistant_status = (rc1 != 0 ? rc1 : rc);
        thread_set_flags(athread, SVC_STOPPED);
-       wake_up_all(&mthread->t_ctl_waitq);
        lad->lad_task = NULL;
        spin_unlock(&lad->lad_lock);
 
@@ -1838,6 +1861,7 @@ fini:
               lad->lad_assistant_status);
 
        lfsck_thread_args_fini(lta);
+       wake_up_all(&mthread->t_ctl_waitq);
 
        return rc;
 }