gfs2: use workqueue instead of slow-work

Workqueue can now handle high concurrency.  Convert gfs to use
workqueue instead of slow-work.

* Steven pointed out that recovery path might be run from allocation
  path and thus requires forward progress guarantee without memory
  allocation.  Create and use gfs_recovery_wq with rescuer.  Please
  note that forward progress wasn't guaranteed with slow-work.

* Updated to use non-reentrant workqueue.

Signed-off-by: Tejun Heo <tj@kernel.org>
Acked-by: Steven Whitehouse <swhiteho@redhat.com>

1 files changed, 21 insertions, 33 deletions

diff --git a/fs/gfs2/recovery.c b/fs/gfs2/recovery.c
index 4b9bece3d437..f7f89a94a5a4 100644
--- a/fs/gfs2/recovery.c
+++ b/fs/gfs2/recovery.c
@@ -14,7 +14,6 @@
 #include <linux/buffer_head.h>
 #include <linux/gfs2_ondisk.h>
 #include <linux/crc32.h>
-#include <linux/slow-work.h>
 
 #include "gfs2.h"
 #include "incore.h"
@@ -28,6 +27,8 @@
 #include "util.h"
 #include "dir.h"
 
+struct workqueue_struct *gfs_recovery_wq;
+
 int gfs2_replay_read_block(struct gfs2_jdesc *jd, unsigned int blk,
                            struct buffer_head **bh)
 {
@@ -443,23 +444,7 @@ static void gfs2_recovery_done(struct gfs2_sbd *sdp, unsigned int jid,
         kobject_uevent_env(&sdp->sd_kobj, KOBJ_CHANGE, envp);
 }
 
-static int gfs2_recover_get_ref(struct slow_work *work)
-{
-        struct gfs2_jdesc *jd = container_of(work, struct gfs2_jdesc, jd_work);
-        if (test_and_set_bit(JDF_RECOVERY, &jd->jd_flags))
-                return -EBUSY;
-        return 0;
-}
-
-static void gfs2_recover_put_ref(struct slow_work *work)
-{
-        struct gfs2_jdesc *jd = container_of(work, struct gfs2_jdesc, jd_work);
-        clear_bit(JDF_RECOVERY, &jd->jd_flags);
-        smp_mb__after_clear_bit();
-        wake_up_bit(&jd->jd_flags, JDF_RECOVERY);
-}
-
-static void gfs2_recover_work(struct slow_work *work)
+void gfs2_recover_func(struct work_struct *work)
 {
         struct gfs2_jdesc *jd = container_of(work, struct gfs2_jdesc, jd_work);
         struct gfs2_inode *ip = GFS2_I(jd->jd_inode);
@@ -578,7 +563,7 @@ static void gfs2_recover_work(struct slow_work *work)
                 gfs2_glock_dq_uninit(&j_gh);
 
         fs_info(sdp, "jid=%u: Done\n", jd->jd_jid);
-        return;
+        goto done;
 
 fail_gunlock_tr:
         gfs2_glock_dq_uninit(&t_gh);
@@ -590,32 +575,35 @@ fail_gunlock_j:
         }
 
         fs_info(sdp, "jid=%u: %s\n", jd->jd_jid, (error) ? "Failed" : "Done");
-
 fail:
         gfs2_recovery_done(sdp, jd->jd_jid, LM_RD_GAVEUP);
+done:
+        clear_bit(JDF_RECOVERY, &jd->jd_flags);
+        smp_mb__after_clear_bit();
+        wake_up_bit(&jd->jd_flags, JDF_RECOVERY);
 }
 
-struct slow_work_ops gfs2_recover_ops = {
-        .owner   = THIS_MODULE,
-        .get_ref = gfs2_recover_get_ref,
-        .put_ref = gfs2_recover_put_ref,
-        .execute = gfs2_recover_work,
-};
-
-
 static int gfs2_recovery_wait(void *word)
 {
         schedule();
         return 0;
 }
 
-int gfs2_recover_journal(struct gfs2_jdesc *jd)
+int gfs2_recover_journal(struct gfs2_jdesc *jd, bool wait)
 {
         int rv;
-        rv = slow_work_enqueue(&jd->jd_work);
-        if (rv)
-                return rv;
-        wait_on_bit(&jd->jd_flags, JDF_RECOVERY, gfs2_recovery_wait, TASK_UNINTERRUPTIBLE);
+
+        if (test_and_set_bit(JDF_RECOVERY, &jd->jd_flags))
+                return -EBUSY;
+
+        /* we have JDF_RECOVERY, queue should always succeed */
+        rv = queue_work(gfs_recovery_wq, &jd->jd_work);
+        BUG_ON(!rv);
+
+        if (wait)
+                wait_on_bit(&jd->jd_flags, JDF_RECOVERY, gfs2_recovery_wait,
+                            TASK_UNINTERRUPTIBLE);
+
         return 0;
 }