dlm: async freeing of lockspace resources

This patch handles freeing of lockspace resources asynchronously besides
the release_lockspace() context. The release_lockspace() context is
sometimes called in a time critical context, e.g. umount syscall. Most
every user space init system will timeout if it takes too long. To
reduce the potential waiting time we deregister in release_lockspace()
the lockspace from the DLM subsystem and do the actual releasing of
lockspace resource in a worker of a workqueue following recommendation
of:

https://lore.kernel.org/all/49925af7-78a8-a3dd-bce6-cfc02e1a9236@I-love.SAKURA.ne.jp/T/#u

as flushing of system workqueues are not allowed. The most time to
release the DLM resources are spent to release the data structures
"ls->ls_lkbxa" and "ls->ls_rsbtbl" as they iterate over each entries and
those data structures can contain millions of entries. This patch handles
for now only freeing of those data structures as those operations are
the most reason why release_lockspace() blocking of being returned.

Signed-off-by: Alexander Aring <aahringo@redhat.com>
Signed-off-by: David Teigland <teigland@redhat.com>

diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h
index 32d98e63d25e2c7d243e2653844b37058f0429c5..0562099e60eba1c03ff1700df108d0bdaf8f6fc4 100644 (file)
--- a/fs/dlm/dlm_internal.h
+++ b/fs/dlm/dlm_internal.h
@@ -660,6 +660,8 @@ struct dlm_ls {
        const struct dlm_lockspace_ops *ls_ops;
        void                    *ls_ops_arg;
 
+       struct work_struct      ls_free_work;
+
        int                     ls_namelen;
        char                    ls_name[DLM_LOCKSPACE_LEN + 1];
 };
@@ -803,6 +805,8 @@ static inline void dlm_set_sbflags_val(struct dlm_lkb *lkb, uint32_t val)
                          __DLM_SBF_MAX_BIT);
 }
 
+extern struct workqueue_struct *dlm_wq;
+
 int dlm_plock_init(void);
 void dlm_plock_exit(void);
 

diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c
index bf14016d53e1bfcfc7cadd16f6c5be46d4f2a7e2..8afac6e2dff0029291102d3c66f09352286ec8b2 100644 (file)
--- a/fs/dlm/lockspace.c
+++ b/fs/dlm/lockspace.c
@@ -315,6 +315,44 @@ static int threads_start(void)
        return error;
 }
 
+static int lkb_idr_free(struct dlm_lkb *lkb)
+{
+       if (lkb->lkb_lvbptr && test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
+               dlm_free_lvb(lkb->lkb_lvbptr);
+
+       dlm_free_lkb(lkb);
+       return 0;
+}
+
+static void rhash_free_rsb(void *ptr, void *arg)
+{
+       struct dlm_rsb *rsb = ptr;
+
+       dlm_free_rsb(rsb);
+}
+
+static void free_lockspace(struct work_struct *work)
+{
+       struct dlm_ls *ls  = container_of(work, struct dlm_ls, ls_free_work);
+       struct dlm_lkb *lkb;
+       unsigned long id;
+
+       /*
+        * Free all lkb's in xa
+        */
+       xa_for_each(&ls->ls_lkbxa, id, lkb) {
+               lkb_idr_free(lkb);
+       }
+       xa_destroy(&ls->ls_lkbxa);
+
+       /*
+        * Free all rsb's on rsbtbl
+        */
+       rhashtable_free_and_destroy(&ls->ls_rsbtbl, rhash_free_rsb, NULL);
+
+       kfree(ls);
+}
+
 static int new_lockspace(const char *name, const char *cluster,
                         uint32_t flags, int lvblen,
                         const struct dlm_lockspace_ops *ops, void *ops_arg,
@@ -445,6 +483,8 @@ static int new_lockspace(const char *name, const char *cluster,
        spin_lock_init(&ls->ls_cb_lock);
        INIT_LIST_HEAD(&ls->ls_cb_delay);
 
+       INIT_WORK(&ls->ls_free_work, free_lockspace);
+
        ls->ls_recoverd_task = NULL;
        mutex_init(&ls->ls_recoverd_active);
        spin_lock_init(&ls->ls_recover_lock);
@@ -627,15 +667,6 @@ int dlm_new_user_lockspace(const char *name, const char *cluster,
                                   ops_arg, ops_result, lockspace);
 }
 
-static int lkb_idr_free(struct dlm_lkb *lkb)
-{
-       if (lkb->lkb_lvbptr && test_bit(DLM_IFL_MSTCPY_BIT, &lkb->lkb_iflags))
-               dlm_free_lvb(lkb->lkb_lvbptr);
-
-       dlm_free_lkb(lkb);
-       return 0;
-}
-
 /* NOTE: We check the lkbxa here rather than the resource table.
    This is because there may be LKBs queued as ASTs that have been unlinked
    from their RSBs and are pending deletion once the AST has been delivered */
@@ -667,17 +698,8 @@ static int lockspace_busy(struct dlm_ls *ls, int force)
        return rv;
 }
 
-static void rhash_free_rsb(void *ptr, void *arg)
-{
-       struct dlm_rsb *rsb = ptr;
-
-       dlm_free_rsb(rsb);
-}
-
 static int release_lockspace(struct dlm_ls *ls, int force)
 {
-       struct dlm_lkb *lkb;
-       unsigned long id;
        int busy, rv;
 
        busy = lockspace_busy(ls, force);
@@ -735,19 +757,6 @@ static int release_lockspace(struct dlm_ls *ls, int force)
        xa_destroy(&ls->ls_recover_xa);
        kfree(ls->ls_recover_buf);
 
-       /*
-        * Free all lkb's in xa
-        */
-       xa_for_each(&ls->ls_lkbxa, id, lkb) {
-               lkb_idr_free(lkb);
-       }
-       xa_destroy(&ls->ls_lkbxa);
-
-       /*
-        * Free all rsb's on rsbtbl
-        */
-       rhashtable_free_and_destroy(&ls->ls_rsbtbl, rhash_free_rsb, NULL);
-
        /*
         * Free structures on any other lists
         */
@@ -757,9 +766,11 @@ static int release_lockspace(struct dlm_ls *ls, int force)
        dlm_clear_members(ls);
        dlm_clear_members_gone(ls);
        kfree(ls->ls_node_array);
-       log_rinfo(ls, "release_lockspace final free");
-       kfree(ls);
 
+       log_rinfo(ls, "%s final free", __func__);
+
+       /* delayed free of data structures see free_lockspace() */
+       queue_work(dlm_wq, &ls->ls_free_work);
        module_put(THIS_MODULE);
        return 0;
 }

diff --git a/fs/dlm/main.c b/fs/dlm/main.c
index 6ca28299c9db182e5387b113e1be26b490508de3..cb15db8ba9bf83a0aa804b927f448cbb3efb5fa9 100644 (file)
--- a/fs/dlm/main.c
+++ b/fs/dlm/main.c
@@ -22,6 +22,8 @@
 #define CREATE_TRACE_POINTS
 #include <trace/events/dlm.h>
 
+struct workqueue_struct *dlm_wq;
+
 static int __init init_dlm(void)
 {
        int error;
@@ -50,10 +52,16 @@ static int __init init_dlm(void)
        if (error)
                goto out_user;
 
+       dlm_wq = alloc_workqueue("dlm_wq", 0, 0);
+       if (!dlm_wq)
+               goto out_plock;
+
        printk("DLM installed\n");
 
        return 0;
 
+ out_plock:
+       dlm_plock_exit();
  out_user:
        dlm_user_exit();
  out_debug:
@@ -70,6 +78,8 @@ static int __init init_dlm(void)
 
 static void __exit exit_dlm(void)
 {
+       /* be sure every pending work e.g. freeing is done */
+       destroy_workqueue(dlm_wq);
        dlm_plock_exit();
        dlm_user_exit();
        dlm_config_exit();