LU-5039 mdd: do orphan cleanup asynchronously

Orphan cleanup shouldn't block mount process, it can be done
asynchronously.

Signed-off-by: Niu Yawei <yawei.niu@intel.com>
Change-Id: Ib282dbece0cf789b33105aaeaeef04bd6f1dcf70
Reviewed-on: http://review.whamcloud.com/10584
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Lai Siyao <lai.siyao@intel.com>
Reviewed-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c
index 2c7a2da..521597d 100644 (file)
--- a/lustre/mdd/mdd_device.c
+++ b/lustre/mdd/mdd_device.c
@@ -786,6 +786,7 @@ static int mdd_hsm_actions_llog_fini(const struct lu_env *env,
 static void mdd_device_shutdown(const struct lu_env *env, struct mdd_device *m,
                                struct lustre_cfg *cfg)
 {
+       mdd_generic_thread_stop(&m->mdd_orph_cleanup_thread);
        lfsck_degister(env, m->mdd_bottom);
        mdd_hsm_actions_llog_fini(env, m);
        mdd_changelog_fini(env, m);
@@ -855,7 +856,7 @@ static int mdd_recovery_complete(const struct lu_env *env,
        next = &mdd->mdd_child->dd_lu_dev;
 
         /* XXX: orphans handling. */
-        __mdd_orphan_cleanup(env, mdd);
+        mdd_orphan_cleanup(env, mdd);
         rc = next->ld_ops->ldo_recovery_complete(env, next);
 
         RETURN(rc);
@@ -1492,6 +1493,35 @@ static void mdd_key_fini(const struct lu_context *ctx,
 /* context key: mdd_thread_key */
 LU_CONTEXT_KEY_DEFINE(mdd, LCT_MD_THREAD);
 
+int mdd_generic_thread_start(struct mdd_generic_thread *thread,
+                            int (*func)(void *), void *data, char *name)
+{
+       struct task_struct      *task;
+
+       LASSERT(thread->mgt_init == false);
+       init_completion(&thread->mgt_started);
+       init_completion(&thread->mgt_finished);
+       thread->mgt_data = data;
+       thread->mgt_abort = false;
+       thread->mgt_init = true;
+
+       task = kthread_run(func, thread, name);
+       if (IS_ERR(task)) {
+               complete(&thread->mgt_finished);
+               return PTR_ERR(task);
+       }
+       wait_for_completion(&thread->mgt_started);
+       return 0;
+}
+
+void mdd_generic_thread_stop(struct mdd_generic_thread *thread)
+{
+       if (thread->mgt_init == true) {
+               thread->mgt_abort = true;
+               wait_for_completion(&thread->mgt_finished);
+       }
+}
+
 static int __init mdd_mod_init(void)
 {
        struct lprocfs_static_vars lvars;

diff --git a/lustre/mdd/mdd_internal.h b/lustre/mdd/mdd_internal.h
index 3353641..59deb41 100644 (file)
--- a/lustre/mdd/mdd_internal.h
+++ b/lustre/mdd/mdd_internal.h
@@ -87,6 +87,14 @@ struct mdd_dot_lustre_objs {
         struct mdd_object *mdd_obf;
 };
 
+struct mdd_generic_thread {
+       struct completion       mgt_started;
+       struct completion       mgt_finished;
+       void                   *mgt_data;
+       bool                    mgt_abort;
+       bool                    mgt_init;
+};
+
 struct mdd_device {
         struct md_device                 mdd_md_dev;
        struct obd_export               *mdd_child_exp;
@@ -104,6 +112,7 @@ struct mdd_device {
        unsigned int                     mdd_sync_permission;
        int                              mdd_connects;
        struct local_oid_storage        *mdd_los;
+       struct mdd_generic_thread        mdd_orph_cleanup_thread;
 };
 
 enum mod_flags {
@@ -328,7 +337,7 @@ struct lu_buf *mdd_buf_get(const struct lu_env *env, void *area, ssize_t len);
 const struct lu_buf *mdd_buf_get_const(const struct lu_env *env,
                                        const void *area, ssize_t len);
 
-int __mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d);
+int mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d);
 int __mdd_orphan_add(const struct lu_env *, struct mdd_object *,
                      struct thandle *);
 int __mdd_orphan_del(const struct lu_env *, struct mdd_object *,
@@ -439,6 +448,9 @@ int mdd_permission(const struct lu_env *env,
                    struct md_attr *ma, int mask);
 int mdd_capa_get(const struct lu_env *env, struct md_object *obj,
                  struct lustre_capa *capa, int renewal);
+int mdd_generic_thread_start(struct mdd_generic_thread *thread,
+                            int (*func)(void *), void *data, char *name);
+void mdd_generic_thread_stop(struct mdd_generic_thread *thread);
 
 /* mdd_prepare.c */
 int mdd_compat_fixes(const struct lu_env *env, struct mdd_device *mdd);

diff --git a/lustre/mdd/mdd_orphans.c b/lustre/mdd/mdd_orphans.c
index 9aa132b..b6ce332 100644 (file)
--- a/lustre/mdd/mdd_orphans.c
+++ b/lustre/mdd/mdd_orphans.c
@@ -419,14 +419,15 @@ static int orph_key_test_and_del(const struct lu_env *env,
  * have to be referenced (opened) by some client during recovery, or they
  * will be deleted here (for clients that did not complete recovery).
  *
- * \param mdd  MDD device finishing recovery
+ * \param thread  info about orphan cleanup thread
  *
  * \retval 0   success
  * \retval -ve error
  */
 static int orph_index_iterate(const struct lu_env *env,
-                             struct mdd_device *mdd)
+                             struct mdd_generic_thread *thread)
 {
+       struct mdd_device *mdd = (struct mdd_device *)thread->mgt_data;
        struct dt_object *dor = mdd->mdd_orphans;
        struct lu_dirent *ent = &mdd_env_info(env)->mti_ent;
        const struct dt_it_ops *iops;
@@ -437,7 +438,6 @@ static int orph_index_iterate(const struct lu_env *env,
         __u64             cookie;
         ENTRY;
 
-        /* In recovery phase, do not need for any lock here */
         iops = &dor->do_index_ops->dio_it;
         it = iops->init(env, dor, LUDA_64BITHASH, BYPASS_CAPA);
         if (IS_ERR(it)) {
@@ -458,6 +458,9 @@ static int orph_index_iterate(const struct lu_env *env,
         }
 
        do {
+               if (thread->mgt_abort)
+                       break;
+
                key_sz = iops->key_size(env, it);
                /* filter out "." and ".." entries from PENDING dir. */
                if (key_sz < 8)
@@ -557,13 +560,59 @@ void orph_index_fini(const struct lu_env *env, struct mdd_device *mdd)
         EXIT;
 }
 
+static int __mdd_orphan_cleanup(void *args)
+{
+       struct mdd_generic_thread *thread = (struct mdd_generic_thread *)args;
+       struct lu_env             *env = NULL;
+       int                        rc;
+       ENTRY;
+
+       complete(&thread->mgt_started);
+
+       OBD_ALLOC_PTR(env);
+       if (env == NULL)
+               GOTO(out, rc = -ENOMEM);
+
+       rc = lu_env_init(env, LCT_MD_THREAD);
+       if (rc)
+               GOTO(out, rc);
+
+       rc = orph_index_iterate(env, thread);
+
+       lu_env_fini(env);
+       GOTO(out, rc);
+out:
+       if (env)
+               OBD_FREE_PTR(env);
+       complete(&thread->mgt_finished);
+       return rc;
+}
+
 /**
  *  Iterate orphan index to cleanup orphan objects after recovery is done.
  *  \param d   mdd device in recovery.
  */
-int __mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
+int mdd_orphan_cleanup(const struct lu_env *env, struct mdd_device *d)
 {
-        return orph_index_iterate(env, d);
+       int rc = -ENOMEM;
+       char *name = NULL;
+
+       OBD_ALLOC(name, MTI_NAME_MAXLEN);
+       if (name == NULL)
+               goto out;
+
+       snprintf(name, MTI_NAME_MAXLEN, "orph_cleanup_%s",
+                mdd2obd_dev(d)->obd_name);
+
+       rc = mdd_generic_thread_start(&d->mdd_orph_cleanup_thread,
+                                     __mdd_orphan_cleanup, (void *)d, name);
+out:
+       if (rc)
+               CERROR("%s: start orphan cleanup thread failed:%d\n",
+                      mdd2obd_dev(d)->obd_name, rc);
+       if (name)
+               OBD_FREE(name, MTI_NAME_MAXLEN);
+       return rc;
 }
 
 /**