LU-12616 obclass: fix MDS start/stop race

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

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index 6d146a0..7e67555 100644 (file)
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -379,6 +379,12 @@ static int __lfsck_ibits_lock(const struct lu_env *env,
                einfo->ei_res_id = resid;
 
                rc = dt_object_lock(env, obj, lh, einfo, policy);
+               /* for regular checks LFSCK doesn't use LDLM locking,
+                * so the state isn't coherent. here we just took LDLM
+                * lock for coherency and it's time to invalidate
+                * previous state */
+               if (rc == ELDLM_OK)
+                       dt_invalidate(env, obj);
        } else {
                rc = ldlm_cli_enqueue_local(env, lfsck->li_namespace, resid,
                                            LDLM_IBITS, policy, mode,
@@ -426,6 +432,36 @@ int lfsck_ibits_lock(const struct lu_env *env, struct lfsck_instance *lfsck,
 }
 
 /**
+ * Request the remote LOOKUP lock for the given object.
+ *
+ * If \a pobj is remote, the LOOKUP lock of \a obj is on the MDT where
+ * \a pobj is, acquire LOOKUP lock there.
+ *
+ * \param[in] env      pointer to the thread context
+ * \param[in] lfsck    pointer to the lfsck instance
+ * \param[in] pobj     pointer to parent dt_object
+ * \param[in] obj      pointer to the dt_object to be locked
+ * \param[out] lh      pointer to the lock handle
+ * \param[in] mode     the mode for the ldlm lock to be acquired
+ *
+ * \retval             0 for success
+ * \retval             negative error number on failure
+ */
+int lfsck_remote_lookup_lock(const struct lu_env *env,
+                            struct lfsck_instance *lfsck,
+                            struct dt_object *pobj, struct dt_object *obj,
+                            struct lustre_handle *lh, enum ldlm_mode mode)
+{
+       struct ldlm_res_id *resid = &lfsck_env_info(env)->lti_resid;
+
+       LASSERT(!lustre_handle_is_used(lh));
+
+       fid_build_reg_res_name(lfsck_dto2fid(obj), resid);
+       return __lfsck_ibits_lock(env, lfsck, pobj, resid, lh,
+                                 MDS_INODELOCK_LOOKUP, mode);
+}
+
+/**
  * Release the the specified ibits lock.
  *
  * If the lock has been acquired before, release it
@@ -1835,14 +1871,12 @@ bool __lfsck_set_speed(struct lfsck_instance *lfsck, __u32 limit)
        bool dirty = false;
 
        if (limit != LFSCK_SPEED_NO_LIMIT) {
-               if (limit > msecs_to_jiffies(MSEC_PER_SEC)) {
-                       lfsck->li_sleep_rate = limit /
-                                              msecs_to_jiffies(MSEC_PER_SEC);
+               if (limit > cfs_time_seconds(1)) {
+                       lfsck->li_sleep_rate = limit / cfs_time_seconds(1);
                        lfsck->li_sleep_jif = 1;
                } else {
                        lfsck->li_sleep_rate = 1;
-                       lfsck->li_sleep_jif = msecs_to_jiffies(MSEC_PER_SEC) /
-                                             limit;
+                       lfsck->li_sleep_jif = cfs_time_seconds(1) / limit;
                }
        } else {
                lfsck->li_sleep_jif = 0;
@@ -2504,10 +2538,7 @@ int lfsck_start_assistant(const struct lu_env *env, struct lfsck_component *com,
 
        lad->lad_assistant_status = 0;
        lad->lad_post_result = 0;
-       lad->lad_to_post = 0;
-       lad->lad_to_double_scan = 0;
-       lad->lad_in_double_scan = 0;
-       lad->lad_exit = 0;
+       lad->lad_flags = 0;
        lad->lad_advance_lock = false;
        thread_set_flags(athread, 0);
 
@@ -2571,8 +2602,8 @@ void lfsck_post_generic(const struct lu_env *env,
 
        lad->lad_post_result = *result;
        if (*result <= 0)
-               lad->lad_exit = 1;
-       lad->lad_to_post = 1;
+               set_bit(LAD_EXIT, &lad->lad_flags);
+       set_bit(LAD_TO_POST, &lad->lad_flags);
 
        CDEBUG(D_LFSCK, "%s: waiting for assistant to do %s post, rc = %d\n",
               lfsck_lfsck2name(com->lc_lfsck), lad->lad_name, *result);
@@ -2599,9 +2630,9 @@ int lfsck_double_scan_generic(const struct lu_env *env,
        struct l_wait_info               lwi     = { 0 };
 
        if (status != LS_SCANNING_PHASE2)
-               lad->lad_exit = 1;
+               set_bit(LAD_EXIT, &lad->lad_flags);
        else
-               lad->lad_to_double_scan = 1;
+               set_bit(LAD_TO_DOUBLE_SCAN, &lad->lad_flags);
 
        CDEBUG(D_LFSCK, "%s: waiting for assistant to do %s double_scan, "
               "status %d\n",
@@ -2609,7 +2640,7 @@ int lfsck_double_scan_generic(const struct lu_env *env,
 
        wake_up_all(&athread->t_ctl_waitq);
        l_wait_event(mthread->t_ctl_waitq,
-                    lad->lad_in_double_scan ||
+                    test_bit(LAD_IN_DOUBLE_SCAN, &lad->lad_flags) ||
                     thread_is_stopped(athread),
                     &lwi);
 
@@ -2631,7 +2662,7 @@ void lfsck_quit_generic(const struct lu_env *env,
        struct ptlrpc_thread            *athread = &lad->lad_thread;
        struct l_wait_info               lwi     = { 0 };
 
-       lad->lad_exit = 1;
+       set_bit(LAD_EXIT, &lad->lad_flags);
        wake_up_all(&athread->t_ctl_waitq);
        l_wait_event(mthread->t_ctl_waitq,
                     thread_is_init(athread) ||
@@ -2739,7 +2770,7 @@ int lfsck_load_sub_trace_files(const struct lu_env *env,
 }
 
 /* external interfaces */
-int lfsck_get_speed(struct seq_file *m, char *buf, struct dt_device *key)
+int lfsck_get_speed(char *buf, struct dt_device *key)
 {
        struct lu_env           env;
        struct lfsck_instance  *lfsck;
@@ -2751,14 +2782,9 @@ int lfsck_get_speed(struct seq_file *m, char *buf, struct dt_device *key)
                RETURN(rc);
 
        lfsck = lfsck_instance_find(key, true, false);
-       if (likely(lfsck != NULL)) {
-               if (m) {
-                       seq_printf(m, "%u\n",
-                                  lfsck->li_bookmark_ram.lb_speed_limit);
-               } else if (buf) {
-                       rc = sprintf(buf, "%u\n",
-                                    lfsck->li_bookmark_ram.lb_speed_limit);
-               }
+       if (lfsck && buf) {
+               rc = sprintf(buf, "%u\n",
+                            lfsck->li_bookmark_ram.lb_speed_limit);
                lfsck_instance_put(&env, lfsck);
        } else {
                rc = -ENXIO;
@@ -2798,7 +2824,7 @@ int lfsck_set_speed(struct dt_device *key, __u32 val)
 }
 EXPORT_SYMBOL(lfsck_set_speed);
 
-int lfsck_get_windows(struct seq_file *m, struct dt_device *key)
+int lfsck_get_windows(char *buf, struct dt_device *key)
 {
        struct lu_env           env;
        struct lfsck_instance  *lfsck;
@@ -2811,7 +2837,8 @@ int lfsck_get_windows(struct seq_file *m, struct dt_device *key)
 
        lfsck = lfsck_instance_find(key, true, false);
        if (likely(lfsck != NULL)) {
-               seq_printf(m, "%u\n", lfsck->li_bookmark_ram.lb_async_windows);
+               rc = sprintf(buf, "%u\n",
+                            lfsck->li_bookmark_ram.lb_async_windows);
                lfsck_instance_put(&env, lfsck);
        } else {
                rc = -ENXIO;
@@ -3046,7 +3073,7 @@ again:
        if (unlikely(rc == -EINPROGRESS)) {
                retry = true;
                set_current_state(TASK_INTERRUPTIBLE);
-               schedule_timeout(msecs_to_jiffies(MSEC_PER_SEC));
+               schedule_timeout(cfs_time_seconds(1));
                set_current_state(TASK_RUNNING);
                if (!signal_pending(current) &&
                    thread_is_running(&lfsck->li_thread))