22 files changed, 262 insertions, 245 deletions

diff --git a/fs/cachefiles/namei.c b/fs/cachefiles/namei.c
index f4a7840bf42c..42c7fafc8bfe 100644
--- a/fs/cachefiles/namei.c
+++ b/fs/cachefiles/namei.c
@@ -37,9 +37,9 @@ void __cachefiles_printk_object(struct cachefiles_object *object,
 
         printk(KERN_ERR "%sobject: OBJ%x\n",
                prefix, object->fscache.debug_id);
-        printk(KERN_ERR "%sobjstate=%s fl=%lx swfl=%lx ev=%lx[%lx]\n",
+        printk(KERN_ERR "%sobjstate=%s fl=%lx wbusy=%x ev=%lx[%lx]\n",
                prefix, fscache_object_states[object->fscache.state],
-               object->fscache.flags, object->fscache.work.flags,
+               object->fscache.flags, work_busy(&object->fscache.work),
                object->fscache.events,
                object->fscache.event_mask & FSCACHE_OBJECT_EVENTS_MASK);
         printk(KERN_ERR "%sops=%u inp=%u exc=%u\n",
@@ -212,7 +212,7 @@ wait_for_old_object:
 
                 /* if the object we're waiting for is queued for processing,
                  * then just put ourselves on the queue behind it */
-                if (slow_work_is_queued(&xobject->fscache.work)) {
+                if (work_pending(&xobject->fscache.work)) {
                         _debug("queue OBJ%x behind OBJ%x immediately",
                                object->fscache.debug_id,
                                xobject->fscache.debug_id);
@@ -220,8 +220,7 @@ wait_for_old_object:
                 }
 
                 /* otherwise we sleep until either the object we're waiting for
-                 * is done, or the slow-work facility wants the thread back to
-                 * do other work */
+                 * is done, or the fscache_object is congested */
                 wq = bit_waitqueue(&xobject->flags, CACHEFILES_OBJECT_ACTIVE);
                 init_wait(&wait);
                 requeue = false;
@@ -229,8 +228,8 @@ wait_for_old_object:
                         prepare_to_wait(wq, &wait, TASK_UNINTERRUPTIBLE);
                         if (!test_bit(CACHEFILES_OBJECT_ACTIVE, &xobject->flags))
                                 break;
-                        requeue = slow_work_sleep_till_thread_needed(
-                                &object->fscache.work, &timeout);
+
+                        requeue = fscache_object_sleep_till_congested(&timeout);
                 } while (timeout > 0 && !requeue);
                 finish_wait(wq, &wait);
 
diff --git a/fs/cachefiles/rdwr.c b/fs/cachefiles/rdwr.c
index 0f0d41fbb03f..0e3c0924cc3a 100644
--- a/fs/cachefiles/rdwr.c
+++ b/fs/cachefiles/rdwr.c
@@ -422,7 +422,7 @@ int cachefiles_read_or_alloc_page(struct fscache_retrieval *op,
         shift = PAGE_SHIFT - inode->i_sb->s_blocksize_bits;
 
         op->op.flags &= FSCACHE_OP_KEEP_FLAGS;
-        op->op.flags |= FSCACHE_OP_FAST;
+        op->op.flags |= FSCACHE_OP_ASYNC;
         op->op.processor = cachefiles_read_copier;
 
         pagevec_init(&pagevec, 0);
@@ -729,7 +729,7 @@ int cachefiles_read_or_alloc_pages(struct fscache_retrieval *op,
         pagevec_init(&pagevec, 0);
 
         op->op.flags &= FSCACHE_OP_KEEP_FLAGS;
-        op->op.flags |= FSCACHE_OP_FAST;
+        op->op.flags |= FSCACHE_OP_ASYNC;
         op->op.processor = cachefiles_read_copier;
 
         INIT_LIST_HEAD(&backpages);
diff --git a/fs/cifs/Kconfig b/fs/cifs/Kconfig
index 5739fd7f88b4..efcb3a78a327 100644
--- a/fs/cifs/Kconfig
+++ b/fs/cifs/Kconfig
@@ -2,7 +2,6 @@ config CIFS
         tristate "CIFS support (advanced network filesystem, SMBFS successor)"
         depends on INET
         select NLS
-        select SLOW_WORK
         help
           This is the client VFS module for the Common Internet File System
           (CIFS) protocol which is the successor to the Server Message Block
diff --git a/fs/cifs/cifsfs.c b/fs/cifs/cifsfs.c
index 8a2cf129e535..5ac541a30c1d 100644
--- a/fs/cifs/cifsfs.c
+++ b/fs/cifs/cifsfs.c
@@ -939,15 +939,10 @@ init_cifs(void)
         if (rc)
                 goto out_unregister_key_type;
 #endif
-        rc = slow_work_register_user(THIS_MODULE);
-        if (rc)
-                goto out_unregister_resolver_key;
 
         return 0;
 
- out_unregister_resolver_key:
 #ifdef CONFIG_CIFS_DFS_UPCALL
-        cifs_exit_dns_resolver();
  out_unregister_key_type:
 #endif
 #ifdef CONFIG_CIFS_UPCALL
diff --git a/fs/cifs/cifsglob.h b/fs/cifs/cifsglob.h
index 59906146ad36..0cdfb8c32ac6 100644
--- a/fs/cifs/cifsglob.h
+++ b/fs/cifs/cifsglob.h
@@ -22,7 +22,7 @@
 #include <linux/in.h>
 #include <linux/in6.h>
 #include <linux/slab.h>
-#include <linux/slow-work.h>
+#include <linux/workqueue.h>
 #include "cifs_fs_sb.h"
 #include "cifsacl.h"
 /*
@@ -356,7 +356,7 @@ struct cifsFileInfo {
         atomic_t count;         /* reference count */
         struct mutex fh_mutex; /* prevents reopen race after dead ses*/
         struct cifs_search_info srch_inf;
-        struct slow_work oplock_break; /* slow_work job for oplock breaks */
+        struct work_struct oplock_break; /* work for oplock breaks */
 };
 
 /* Take a reference on the file private data */
@@ -728,6 +728,10 @@ GLOBAL_EXTERN unsigned int cifs_min_rcv;    /* min size of big ntwrk buf pool */
 GLOBAL_EXTERN unsigned int cifs_min_small;  /* min size of small buf pool */
 GLOBAL_EXTERN unsigned int cifs_max_pending; /* MAX requests at once to server*/
 
+void cifs_oplock_break(struct work_struct *work);
+void cifs_oplock_break_get(struct cifsFileInfo *cfile);
+void cifs_oplock_break_put(struct cifsFileInfo *cfile);
+
 extern const struct slow_work_ops cifs_oplock_break_ops;
 
 #endif  /* _CIFS_GLOB_H */
diff --git a/fs/cifs/dir.c b/fs/cifs/dir.c
index a7de5e9fff11..578d88c5b46e 100644
--- a/fs/cifs/dir.c
+++ b/fs/cifs/dir.c
@@ -157,7 +157,7 @@ cifs_new_fileinfo(struct inode *newinode, __u16 fileHandle,
         mutex_init(&pCifsFile->lock_mutex);
         INIT_LIST_HEAD(&pCifsFile->llist);
         atomic_set(&pCifsFile->count, 1);
-        slow_work_init(&pCifsFile->oplock_break, &cifs_oplock_break_ops);
+        INIT_WORK(&pCifsFile->oplock_break, cifs_oplock_break);
 
         write_lock(&GlobalSMBSeslock);
         list_add(&pCifsFile->tlist, &cifs_sb->tcon->openFileList);
diff --git a/fs/cifs/file.c b/fs/cifs/file.c
index fa04a00d126d..db11fdef0e92 100644
--- a/fs/cifs/file.c
+++ b/fs/cifs/file.c
@@ -2307,8 +2307,7 @@ static void cifs_invalidate_page(struct page *page, unsigned long offset)
                 cifs_fscache_invalidate_page(page, &cifsi->vfs_inode);
 }
 
-static void
-cifs_oplock_break(struct slow_work *work)
+void cifs_oplock_break(struct work_struct *work)
 {
         struct cifsFileInfo *cfile = container_of(work, struct cifsFileInfo,
                                                   oplock_break);
@@ -2345,33 +2344,30 @@ cifs_oplock_break(struct slow_work *work)
                                  LOCKING_ANDX_OPLOCK_RELEASE, false);
                 cFYI(1, "Oplock release rc = %d", rc);
         }
+
+        /*
+         * We might have kicked in before is_valid_oplock_break()
+         * finished grabbing reference for us.  Make sure it's done by
+         * waiting for GlobalSMSSeslock.
+         */
+        write_lock(&GlobalSMBSeslock);
+        write_unlock(&GlobalSMBSeslock);
+
+        cifs_oplock_break_put(cfile);
 }
 
-static int
-cifs_oplock_break_get(struct slow_work *work)
+void cifs_oplock_break_get(struct cifsFileInfo *cfile)
 {
-        struct cifsFileInfo *cfile = container_of(work, struct cifsFileInfo,
-                                                  oplock_break);
         mntget(cfile->mnt);
         cifsFileInfo_get(cfile);
-        return 0;
 }
 
-static void
-cifs_oplock_break_put(struct slow_work *work)
+void cifs_oplock_break_put(struct cifsFileInfo *cfile)
 {
-        struct cifsFileInfo *cfile = container_of(work, struct cifsFileInfo,
-                                                  oplock_break);
         mntput(cfile->mnt);
         cifsFileInfo_put(cfile);
 }
 
-const struct slow_work_ops cifs_oplock_break_ops = {
-        .get_ref        = cifs_oplock_break_get,
-        .put_ref        = cifs_oplock_break_put,
-        .execute        = cifs_oplock_break,
-};
-
 const struct address_space_operations cifs_addr_ops = {
         .readpage = cifs_readpage,
         .readpages = cifs_readpages,
diff --git a/fs/cifs/misc.c b/fs/cifs/misc.c
index 1394aa37f26c..3ccadc1326d6 100644
--- a/fs/cifs/misc.c
+++ b/fs/cifs/misc.c
@@ -498,7 +498,6 @@ is_valid_oplock_break(struct smb_hdr *buf, struct TCP_Server_Info *srv)
         struct cifsTconInfo *tcon;
         struct cifsInodeInfo *pCifsInode;
         struct cifsFileInfo *netfile;
-        int rc;
 
         cFYI(1, "Checking for oplock break or dnotify response");
         if ((pSMB->hdr.Command == SMB_COM_NT_TRANSACT) &&
@@ -583,13 +582,18 @@ is_valid_oplock_break(struct smb_hdr *buf, struct TCP_Server_Info *srv)
                                 pCifsInode->clientCanCacheAll = false;
                                 if (pSMB->OplockLevel == 0)
                                         pCifsInode->clientCanCacheRead = false;
-                                rc = slow_work_enqueue(&netfile->oplock_break);
-                                if (rc) {
-                                        cERROR(1, "failed to enqueue oplock "
-                                                   "break: %d\n", rc);
-                                } else {
-                                        netfile->oplock_break_cancelled = false;
-                                }
+
+                                /*
+                                 * cifs_oplock_break_put() can't be called
+                                 * from here.  Get reference after queueing
+                                 * succeeded.  cifs_oplock_break() will
+                                 * synchronize using GlobalSMSSeslock.
+                                 */
+                                if (queue_work(system_nrt_wq,
+                                               &netfile->oplock_break))
+                                        cifs_oplock_break_get(netfile);
+                                netfile->oplock_break_cancelled = false;
+
                                 read_unlock(&GlobalSMBSeslock);
                                 read_unlock(&cifs_tcp_ses_lock);
                                 return true;
diff --git a/fs/fscache/Kconfig b/fs/fscache/Kconfig
index cc94bb9563f2..3f6dfa989881 100644
--- a/fs/fscache/Kconfig
+++ b/fs/fscache/Kconfig
@@ -1,7 +1,6 @@
 
 config FSCACHE
         tristate "General filesystem local caching manager"
-        select SLOW_WORK
         help
           This option enables a generic filesystem caching manager that can be
           used by various network and other filesystems to cache data locally.
diff --git a/fs/fscache/internal.h b/fs/fscache/internal.h
index edd7434ab6e5..6a026441c5a6 100644
--- a/fs/fscache/internal.h
+++ b/fs/fscache/internal.h
@@ -82,6 +82,14 @@ extern unsigned fscache_defer_lookup;
 extern unsigned fscache_defer_create;
 extern unsigned fscache_debug;
 extern struct kobject *fscache_root;
+extern struct workqueue_struct *fscache_object_wq;
+extern struct workqueue_struct *fscache_op_wq;
+DECLARE_PER_CPU(wait_queue_head_t, fscache_object_cong_wait);
+
+static inline bool fscache_object_congested(void)
+{
+        return workqueue_congested(WORK_CPU_UNBOUND, fscache_object_wq);
+}
 
 extern int fscache_wait_bit(void *);
 extern int fscache_wait_bit_interruptible(void *);
diff --git a/fs/fscache/main.c b/fs/fscache/main.c
index add6bdb53f04..f9d856773f79 100644
--- a/fs/fscache/main.c
+++ b/fs/fscache/main.c
@@ -15,6 +15,7 @@
 #include <linux/sched.h>
 #include <linux/completion.h>
 #include <linux/slab.h>
+#include <linux/seq_file.h>
 #include "internal.h"
 
 MODULE_DESCRIPTION("FS Cache Manager");
@@ -40,22 +41,105 @@ MODULE_PARM_DESC(fscache_debug,
                  "FS-Cache debugging mask");
 
 struct kobject *fscache_root;
+struct workqueue_struct *fscache_object_wq;
+struct workqueue_struct *fscache_op_wq;
+
+DEFINE_PER_CPU(wait_queue_head_t, fscache_object_cong_wait);
+
+/* these values serve as lower bounds, will be adjusted in fscache_init() */
+static unsigned fscache_object_max_active = 4;
+static unsigned fscache_op_max_active = 2;
+
+#ifdef CONFIG_SYSCTL
+static struct ctl_table_header *fscache_sysctl_header;
+
+static int fscache_max_active_sysctl(struct ctl_table *table, int write,
+                                     void __user *buffer,
+                                     size_t *lenp, loff_t *ppos)
+{
+        struct workqueue_struct **wqp = table->extra1;
+        unsigned int *datap = table->data;
+        int ret;
+
+        ret = proc_dointvec(table, write, buffer, lenp, ppos);
+        if (ret == 0)
+                workqueue_set_max_active(*wqp, *datap);
+        return ret;
+}
+
+ctl_table fscache_sysctls[] = {
+        {
+                .procname       = "object_max_active",
+                .data           = &fscache_object_max_active,
+                .maxlen         = sizeof(unsigned),
+                .mode           = 0644,
+                .proc_handler   = fscache_max_active_sysctl,
+                .extra1         = &fscache_object_wq,
+        },
+        {
+                .procname       = "operation_max_active",
+                .data           = &fscache_op_max_active,
+                .maxlen         = sizeof(unsigned),
+                .mode           = 0644,
+                .proc_handler   = fscache_max_active_sysctl,
+                .extra1         = &fscache_op_wq,
+        },
+        {}
+};
+
+ctl_table fscache_sysctls_root[] = {
+        {
+                .procname       = "fscache",
+                .mode           = 0555,
+                .child          = fscache_sysctls,
+        },
+        {}
+};
+#endif
 
 /*
  * initialise the fs caching module
  */
 static int __init fscache_init(void)
 {
+        unsigned int nr_cpus = num_possible_cpus();
+        unsigned int cpu;
         int ret;
 
-        ret = slow_work_register_user(THIS_MODULE);
-        if (ret < 0)
-                goto error_slow_work;
+        fscache_object_max_active =
+                clamp_val(nr_cpus,
+                          fscache_object_max_active, WQ_UNBOUND_MAX_ACTIVE);
+
+        ret = -ENOMEM;
+        fscache_object_wq = alloc_workqueue("fscache_object", WQ_UNBOUND,
+                                            fscache_object_max_active);
+        if (!fscache_object_wq)
+                goto error_object_wq;
+
+        fscache_op_max_active =
+                clamp_val(fscache_object_max_active / 2,
+                          fscache_op_max_active, WQ_UNBOUND_MAX_ACTIVE);
+
+        ret = -ENOMEM;
+        fscache_op_wq = alloc_workqueue("fscache_operation", WQ_UNBOUND,
+                                        fscache_op_max_active);
+        if (!fscache_op_wq)
+                goto error_op_wq;
+
+        for_each_possible_cpu(cpu)
+                init_waitqueue_head(&per_cpu(fscache_object_cong_wait, cpu));
 
         ret = fscache_proc_init();
         if (ret < 0)
                 goto error_proc;
 
+#ifdef CONFIG_SYSCTL
+        ret = -ENOMEM;
+        fscache_sysctl_header = register_sysctl_table(fscache_sysctls_root);
+        if (!fscache_sysctl_header)
+                goto error_sysctl;
+#endif
+
         fscache_cookie_jar = kmem_cache_create("fscache_cookie_jar",
                                                sizeof(struct fscache_cookie),
                                                0,
@@ -78,10 +162,16 @@ static int __init fscache_init(void)
 error_kobj:
         kmem_cache_destroy(fscache_cookie_jar);
 error_cookie_jar:
+#ifdef CONFIG_SYSCTL
+        unregister_sysctl_table(fscache_sysctl_header);
+error_sysctl:
+#endif
         fscache_proc_cleanup();
 error_proc:
-        slow_work_unregister_user(THIS_MODULE);
-error_slow_work:
+        destroy_workqueue(fscache_op_wq);
+error_op_wq:
+        destroy_workqueue(fscache_object_wq);
+error_object_wq:
         return ret;
 }
 
@@ -96,8 +186,12 @@ static void __exit fscache_exit(void)
 
         kobject_put(fscache_root);
         kmem_cache_destroy(fscache_cookie_jar);
+#ifdef CONFIG_SYSCTL
+        unregister_sysctl_table(fscache_sysctl_header);
+#endif
         fscache_proc_cleanup();
-        slow_work_unregister_user(THIS_MODULE);
+        destroy_workqueue(fscache_op_wq);
+        destroy_workqueue(fscache_object_wq);
         printk(KERN_NOTICE "FS-Cache: Unloaded\n");
 }
 
diff --git a/fs/fscache/object-list.c b/fs/fscache/object-list.c
index 4a8eb31c5338..ebe29c581380 100644
--- a/fs/fscache/object-list.c
+++ b/fs/fscache/object-list.c
@@ -34,8 +34,8 @@ struct fscache_objlist_data {
 #define FSCACHE_OBJLIST_CONFIG_NOREADS  0x00000200      /* show objects without active reads */
 #define FSCACHE_OBJLIST_CONFIG_EVENTS   0x00000400      /* show objects with events */
 #define FSCACHE_OBJLIST_CONFIG_NOEVENTS 0x00000800      /* show objects without no events */
-#define FSCACHE_OBJLIST_CONFIG_WORK     0x00001000      /* show objects with slow work */
-#define FSCACHE_OBJLIST_CONFIG_NOWORK   0x00002000      /* show objects without slow work */
+#define FSCACHE_OBJLIST_CONFIG_WORK     0x00001000      /* show objects with work */
+#define FSCACHE_OBJLIST_CONFIG_NOWORK   0x00002000      /* show objects without work */
 
         u8              buf[512];       /* key and aux data buffer */
 };
@@ -231,12 +231,11 @@ static int fscache_objlist_show(struct seq_file *m, void *v)
                        READS, NOREADS);
                 FILTER(obj->events & obj->event_mask,
                        EVENTS, NOEVENTS);
-                FILTER(obj->work.flags & ~(1UL << SLOW_WORK_VERY_SLOW),
-                       WORK, NOWORK);
+                FILTER(work_busy(&obj->work), WORK, NOWORK);
         }
 
         seq_printf(m,
-                   "%8x %8x %s %5u %3u %3u %3u %2u %5u %2lx %2lx %1lx %1lx | ",
+                   "%8x %8x %s %5u %3u %3u %3u %2u %5u %2lx %2lx %1lx %1x | ",
                    obj->debug_id,
                    obj->parent ? obj->parent->debug_id : -1,
                    fscache_object_states_short[obj->state],
@@ -249,7 +248,7 @@ static int fscache_objlist_show(struct seq_file *m, void *v)
                    obj->event_mask & FSCACHE_OBJECT_EVENTS_MASK,
                    obj->events,
                    obj->flags,
-                   obj->work.flags);
+                   work_busy(&obj->work));
 
         no_cookie = true;
         keylen = auxlen = 0;
diff --git a/fs/fscache/object.c b/fs/fscache/object.c
index 0b589a9b4ffc..b6b897c550ac 100644
--- a/fs/fscache/object.c
+++ b/fs/fscache/object.c
@@ -14,7 +14,6 @@
 
 #define FSCACHE_DEBUG_LEVEL COOKIE
 #include <linux/module.h>
-#include <linux/seq_file.h>
 #include "internal.h"
 
 const char *fscache_object_states[FSCACHE_OBJECT__NSTATES] = {
@@ -50,12 +49,8 @@ const char fscache_object_states_short[FSCACHE_OBJECT__NSTATES][5] = {
         [FSCACHE_OBJECT_DEAD]           = "DEAD",
 };
 
-static void fscache_object_slow_work_put_ref(struct slow_work *);
-static int  fscache_object_slow_work_get_ref(struct slow_work *);
-static void fscache_object_slow_work_execute(struct slow_work *);
-#ifdef CONFIG_SLOW_WORK_DEBUG
-static void fscache_object_slow_work_desc(struct slow_work *, struct seq_file *);
-#endif
+static int  fscache_get_object(struct fscache_object *);
+static void fscache_put_object(struct fscache_object *);
 static void fscache_initialise_object(struct fscache_object *);
 static void fscache_lookup_object(struct fscache_object *);
 static void fscache_object_available(struct fscache_object *);
@@ -64,17 +59,6 @@ static void fscache_withdraw_object(struct fscache_object *);
 static void fscache_enqueue_dependents(struct fscache_object *);
 static void fscache_dequeue_object(struct fscache_object *);
 
-const struct slow_work_ops fscache_object_slow_work_ops = {
-        .owner          = THIS_MODULE,
-        .get_ref        = fscache_object_slow_work_get_ref,
-        .put_ref        = fscache_object_slow_work_put_ref,
-        .execute        = fscache_object_slow_work_execute,
-#ifdef CONFIG_SLOW_WORK_DEBUG
-        .desc           = fscache_object_slow_work_desc,
-#endif
-};
-EXPORT_SYMBOL(fscache_object_slow_work_ops);
-
 /*
  * we need to notify the parent when an op completes that we had outstanding
  * upon it
@@ -345,7 +329,7 @@ unsupported_event:
 /*
  * execute an object
  */
-static void fscache_object_slow_work_execute(struct slow_work *work)
+void fscache_object_work_func(struct work_struct *work)
 {
         struct fscache_object *object =
                 container_of(work, struct fscache_object, work);
@@ -359,23 +343,9 @@ static void fscache_object_slow_work_execute(struct slow_work *work)
         if (object->events & object->event_mask)
                 fscache_enqueue_object(object);
         clear_bit(FSCACHE_OBJECT_EV_REQUEUE, &object->events);
+        fscache_put_object(object);
 }
-
-/*
- * describe an object for slow-work debugging
- */
-#ifdef CONFIG_SLOW_WORK_DEBUG
-static void fscache_object_slow_work_desc(struct slow_work *work,
-                                          struct seq_file *m)
-{
-        struct fscache_object *object =
-                container_of(work, struct fscache_object, work);
-
-        seq_printf(m, "FSC: OBJ%x: %s",
-                   object->debug_id,
-                   fscache_object_states_short[object->state]);
-}
-#endif
+EXPORT_SYMBOL(fscache_object_work_func);
 
 /*
  * initialise an object
@@ -393,7 +363,6 @@ static void fscache_initialise_object(struct fscache_object *object)
         _enter("");
         ASSERT(object->cookie != NULL);
         ASSERT(object->cookie->parent != NULL);
-        ASSERT(list_empty(&object->work.link));
 
         if (object->events & ((1 << FSCACHE_OBJECT_EV_ERROR) |
                               (1 << FSCACHE_OBJECT_EV_RELEASE) |
@@ -671,10 +640,8 @@ static void fscache_drop_object(struct fscache_object *object)
                 object->parent = NULL;
         }
 
-        /* this just shifts the object release to the slow work processor */
-        fscache_stat(&fscache_n_cop_put_object);
-        object->cache->ops->put_object(object);
-        fscache_stat_d(&fscache_n_cop_put_object);
+        /* this just shifts the object release to the work processor */
+        fscache_put_object(object);
 
         _leave("");
 }
@@ -758,12 +725,10 @@ void fscache_withdrawing_object(struct fscache_cache *cache,
 }
 
 /*
- * allow the slow work item processor to get a ref on an object
+ * get a ref on an object
  */
-static int fscache_object_slow_work_get_ref(struct slow_work *work)
+static int fscache_get_object(struct fscache_object *object)
 {
-        struct fscache_object *object =
-                container_of(work, struct fscache_object, work);
         int ret;
 
         fscache_stat(&fscache_n_cop_grab_object);
@@ -773,13 +738,10 @@ static int fscache_object_slow_work_get_ref(struct slow_work *work)
 }
 
 /*
- * allow the slow work item processor to discard a ref on a work item
+ * discard a ref on a work item
  */
-static void fscache_object_slow_work_put_ref(struct slow_work *work)
+static void fscache_put_object(struct fscache_object *object)
 {
-        struct fscache_object *object =
-                container_of(work, struct fscache_object, work);
-
         fscache_stat(&fscache_n_cop_put_object);
         object->cache->ops->put_object(object);
         fscache_stat_d(&fscache_n_cop_put_object);
@@ -792,8 +754,48 @@ void fscache_enqueue_object(struct fscache_object *object)
 {
         _enter("{OBJ%x}", object->debug_id);
 
-        slow_work_enqueue(&object->work);
+        if (fscache_get_object(object) >= 0) {
+                wait_queue_head_t *cong_wq =
+                        &get_cpu_var(fscache_object_cong_wait);
+
+                if (queue_work(fscache_object_wq, &object->work)) {
+                        if (fscache_object_congested())
+                                wake_up(cong_wq);
+                } else
+                        fscache_put_object(object);
+
+                put_cpu_var(fscache_object_cong_wait);
+        }
+}
+
+/**
+ * fscache_object_sleep_till_congested - Sleep until object wq is congested
+ * @timoutp: Scheduler sleep timeout
+ *
+ * Allow an object handler to sleep until the object workqueue is congested.
+ *
+ * The caller must set up a wake up event before calling this and must have set
+ * the appropriate sleep mode (such as TASK_UNINTERRUPTIBLE) and tested its own
+ * condition before calling this function as no test is made here.
+ *
+ * %true is returned if the object wq is congested, %false otherwise.
+ */
+bool fscache_object_sleep_till_congested(signed long *timeoutp)
+{
+        wait_queue_head_t *cong_wq = &__get_cpu_var(fscache_object_cong_wait);
+        DEFINE_WAIT(wait);
+
+        if (fscache_object_congested())
+                return true;
+
+        add_wait_queue_exclusive(cong_wq, &wait);
+        if (!fscache_object_congested())
+                *timeoutp = schedule_timeout(*timeoutp);
+        finish_wait(cong_wq, &wait);
+
+        return fscache_object_congested();
 }
+EXPORT_SYMBOL_GPL(fscache_object_sleep_till_congested);
 
 /*
  * enqueue the dependents of an object for metadata-type processing
@@ -819,9 +821,7 @@ static void fscache_enqueue_dependents(struct fscache_object *object)
 
                 /* sort onto appropriate lists */
                 fscache_enqueue_object(dep);
-                fscache_stat(&fscache_n_cop_put_object);
-                dep->cache->ops->put_object(dep);
-                fscache_stat_d(&fscache_n_cop_put_object);
+                fscache_put_object(dep);
 
                 if (!list_empty(&object->dependents))
                         cond_resched_lock(&object->lock);
diff --git a/fs/fscache/operation.c b/fs/fscache/operation.c
index f17cecafae44..b9f34eaede09 100644
--- a/fs/fscache/operation.c
+++ b/fs/fscache/operation.c
@@ -42,16 +42,12 @@ void fscache_enqueue_operation(struct fscache_operation *op)
 
         fscache_stat(&fscache_n_op_enqueue);
         switch (op->flags & FSCACHE_OP_TYPE) {
-        case FSCACHE_OP_FAST:
-                _debug("queue fast");
+        case FSCACHE_OP_ASYNC:
+                _debug("queue async");
                 atomic_inc(&op->usage);
-                if (!schedule_work(&op->fast_work))
+                if (!queue_work(fscache_op_wq, &op->work))
                         fscache_put_operation(op);
                 break;
-        case FSCACHE_OP_SLOW:
-                _debug("queue slow");
-                slow_work_enqueue(&op->slow_work);
-                break;
         case FSCACHE_OP_MYTHREAD:
                 _debug("queue for caller's attention");
                 break;
@@ -455,36 +451,13 @@ void fscache_operation_gc(struct work_struct *work)
 }
 
 /*
- * allow the slow work item processor to get a ref on an operation
- */
-static int fscache_op_get_ref(struct slow_work *work)
-{
-        struct fscache_operation *op =
-                container_of(work, struct fscache_operation, slow_work);
-
-        atomic_inc(&op->usage);
-        return 0;
-}
-
-/*
- * allow the slow work item processor to discard a ref on an operation
- */
-static void fscache_op_put_ref(struct slow_work *work)
-{
-        struct fscache_operation *op =
-                container_of(work, struct fscache_operation, slow_work);
-
-        fscache_put_operation(op);
-}
-
-/*
- * execute an operation using the slow thread pool to provide processing context
- * - the caller holds a ref to this object, so we don't need to hold one
+ * execute an operation using fs_op_wq to provide processing context -
+ * the caller holds a ref to this object, so we don't need to hold one
  */
-static void fscache_op_execute(struct slow_work *work)
+void fscache_op_work_func(struct work_struct *work)
 {
         struct fscache_operation *op =
-                container_of(work, struct fscache_operation, slow_work);
+                container_of(work, struct fscache_operation, work);
         unsigned long start;
 
         _enter("{OBJ%x OP%x,%d}",
@@ -494,31 +467,7 @@ static void fscache_op_execute(struct slow_work *work)
         start = jiffies;
         op->processor(op);
         fscache_hist(fscache_ops_histogram, start);
+        fscache_put_operation(op);
 
         _leave("");
 }
-
-/*
- * describe an operation for slow-work debugging
- */
-#ifdef CONFIG_SLOW_WORK_DEBUG
-static void fscache_op_desc(struct slow_work *work, struct seq_file *m)
-{
-        struct fscache_operation *op =
-                container_of(work, struct fscache_operation, slow_work);
-
-        seq_printf(m, "FSC: OBJ%x OP%x: %s/%s fl=%lx",
-                   op->object->debug_id, op->debug_id,
-                   op->name, op->state, op->flags);
-}
-#endif
-
-const struct slow_work_ops fscache_op_slow_work_ops = {
-        .owner          = THIS_MODULE,
-        .get_ref        = fscache_op_get_ref,
-        .put_ref        = fscache_op_put_ref,
-        .execute        = fscache_op_execute,
-#ifdef CONFIG_SLOW_WORK_DEBUG
-        .desc           = fscache_op_desc,
-#endif
-};
diff --git a/fs/fscache/page.c b/fs/fscache/page.c
index 723b889fd219..41c441c2058d 100644
--- a/fs/fscache/page.c
+++ b/fs/fscache/page.c
@@ -105,7 +105,7 @@ bool __fscache_maybe_release_page(struct fscache_cookie *cookie,
 
 page_busy:
         /* we might want to wait here, but that could deadlock the allocator as
-         * the slow-work threads writing to the cache may all end up sleeping
+         * the work threads writing to the cache may all end up sleeping
          * on memory allocation */
         fscache_stat(&fscache_n_store_vmscan_busy);
         return false;
@@ -188,9 +188,8 @@ int __fscache_attr_changed(struct fscache_cookie *cookie)
                 return -ENOMEM;
         }
 
-        fscache_operation_init(op, NULL);
-        fscache_operation_init_slow(op, fscache_attr_changed_op);
-        op->flags = FSCACHE_OP_SLOW | (1 << FSCACHE_OP_EXCLUSIVE);
+        fscache_operation_init(op, fscache_attr_changed_op, NULL);
+        op->flags = FSCACHE_OP_ASYNC | (1 << FSCACHE_OP_EXCLUSIVE);
         fscache_set_op_name(op, "Attr");
 
         spin_lock(&cookie->lock);
@@ -218,24 +217,6 @@ nobufs:
 EXPORT_SYMBOL(__fscache_attr_changed);
 
 /*
- * handle secondary execution given to a retrieval op on behalf of the
- * cache
- */
-static void fscache_retrieval_work(struct work_struct *work)
-{
-        struct fscache_retrieval *op =
-                container_of(work, struct fscache_retrieval, op.fast_work);
-        unsigned long start;
-
-        _enter("{OP%x}", op->op.debug_id);
-
-        start = jiffies;
-        op->op.processor(&op->op);
-        fscache_hist(fscache_ops_histogram, start);
-        fscache_put_operation(&op->op);
-}
-
-/*
  * release a retrieval op reference
  */
 static void fscache_release_retrieval_op(struct fscache_operation *_op)
@@ -269,13 +250,12 @@ static struct fscache_retrieval *fscache_alloc_retrieval(
                 return NULL;
         }
 
-        fscache_operation_init(&op->op, fscache_release_retrieval_op);
+        fscache_operation_init(&op->op, NULL, fscache_release_retrieval_op);
         op->op.flags    = FSCACHE_OP_MYTHREAD | (1 << FSCACHE_OP_WAITING);
         op->mapping     = mapping;
         op->end_io_func = end_io_func;
         op->context     = context;
         op->start_time  = jiffies;
-        INIT_WORK(&op->op.fast_work, fscache_retrieval_work);
         INIT_LIST_HEAD(&op->to_do);
         fscache_set_op_name(&op->op, "Retr");
         return op;
@@ -795,9 +775,9 @@ int __fscache_write_page(struct fscache_cookie *cookie,
         if (!op)
                 goto nomem;
 
-        fscache_operation_init(&op->op, fscache_release_write_op);
-        fscache_operation_init_slow(&op->op, fscache_write_op);
-        op->op.flags = FSCACHE_OP_SLOW | (1 << FSCACHE_OP_WAITING);
+        fscache_operation_init(&op->op, fscache_write_op,
+                               fscache_release_write_op);
+        op->op.flags = FSCACHE_OP_ASYNC | (1 << FSCACHE_OP_WAITING);
         fscache_set_op_name(&op->op, "Write1");
 
         ret = radix_tree_preload(gfp & ~__GFP_HIGHMEM);
@@ -852,7 +832,7 @@ int __fscache_write_page(struct fscache_cookie *cookie,
         fscache_stat(&fscache_n_store_ops);
         fscache_stat(&fscache_n_stores_ok);
 
-        /* the slow work queue now carries its own ref on the object */
+        /* the work queue now carries its own ref on the object */
         fscache_put_operation(&op->op);
         _leave(" = 0");
         return 0;
diff --git a/fs/gfs2/Kconfig b/fs/gfs2/Kconfig
index a47b43107112..cc9665522148 100644
--- a/fs/gfs2/Kconfig
+++ b/fs/gfs2/Kconfig
@@ -7,7 +7,6 @@ config GFS2_FS
         select IP_SCTP if DLM_SCTP
         select FS_POSIX_ACL
         select CRC32
-        select SLOW_WORK
         select QUOTACTL
         help
           A cluster filesystem.
diff --git a/fs/gfs2/incore.h b/fs/gfs2/incore.h
index 8fcbce48a128..fdbf4b366fa5 100644
--- a/fs/gfs2/incore.h
+++ b/fs/gfs2/incore.h
@@ -12,7 +12,6 @@
 
 #include <linux/fs.h>
 #include <linux/workqueue.h>
-#include <linux/slow-work.h>
 #include <linux/dlm.h>
 #include <linux/buffer_head.h>
 
@@ -383,7 +382,7 @@ struct gfs2_journal_extent {
 struct gfs2_jdesc {
         struct list_head jd_list;
         struct list_head extent_list;
-        struct slow_work jd_work;
+        struct work_struct jd_work;
         struct inode *jd_inode;
         unsigned long jd_flags;
 #define JDF_RECOVERY 1
diff --git a/fs/gfs2/main.c b/fs/gfs2/main.c
index fb2a5f93b7c3..b1e9630eb46a 100644
--- a/fs/gfs2/main.c
+++ b/fs/gfs2/main.c
@@ -15,7 +15,6 @@
 #include <linux/init.h>
 #include <linux/gfs2_ondisk.h>
 #include <asm/atomic.h>
-#include <linux/slow-work.h>
 
 #include "gfs2.h"
 #include "incore.h"
@@ -24,6 +23,7 @@
 #include "util.h"
 #include "glock.h"
 #include "quota.h"
+#include "recovery.h"
 
 static struct shrinker qd_shrinker = {
         .shrink = gfs2_shrink_qd_memory,
@@ -138,9 +138,11 @@ static int __init init_gfs2_fs(void)
         if (error)
                 goto fail_unregister;
 
-        error = slow_work_register_user(THIS_MODULE);
-        if (error)
-                goto fail_slow;
+        error = -ENOMEM;
+        gfs_recovery_wq = alloc_workqueue("gfs_recovery",
+                                          WQ_NON_REENTRANT | WQ_RESCUER, 0);
+        if (!gfs_recovery_wq)
+                goto fail_wq;
 
         gfs2_register_debugfs();
 
@@ -148,7 +150,7 @@ static int __init init_gfs2_fs(void)
 
         return 0;
 
-fail_slow:
+fail_wq:
         unregister_filesystem(&gfs2meta_fs_type);
 fail_unregister:
         unregister_filesystem(&gfs2_fs_type);
@@ -190,7 +192,7 @@ static void __exit exit_gfs2_fs(void)
         gfs2_unregister_debugfs();
         unregister_filesystem(&gfs2_fs_type);
         unregister_filesystem(&gfs2meta_fs_type);
-        slow_work_unregister_user(THIS_MODULE);
+        destroy_workqueue(gfs_recovery_wq);
 
         kmem_cache_destroy(gfs2_quotad_cachep);
         kmem_cache_destroy(gfs2_rgrpd_cachep);
diff --git a/fs/gfs2/ops_fstype.c b/fs/gfs2/ops_fstype.c
index 45a4a36195d8..4f44bdeb2f03 100644
--- a/fs/gfs2/ops_fstype.c
+++ b/fs/gfs2/ops_fstype.c
@@ -17,7 +17,6 @@
 #include <linux/namei.h>
 #include <linux/mount.h>
 #include <linux/gfs2_ondisk.h>
-#include <linux/slow-work.h>
 #include <linux/quotaops.h>
 
 #include "gfs2.h"
@@ -673,7 +672,7 @@ static int gfs2_jindex_hold(struct gfs2_sbd *sdp, struct gfs2_holder *ji_gh)
                         break;
 
                 INIT_LIST_HEAD(&jd->extent_list);
-                slow_work_init(&jd->jd_work, &gfs2_recover_ops);
+                INIT_WORK(&jd->jd_work, gfs2_recover_func);
                 jd->jd_inode = gfs2_lookupi(sdp->sd_jindex, &name, 1);
                 if (!jd->jd_inode || IS_ERR(jd->jd_inode)) {
                         if (!jd->jd_inode)
@@ -782,7 +781,8 @@ static int init_journal(struct gfs2_sbd *sdp, int undo)
         if (sdp->sd_lockstruct.ls_first) {
                 unsigned int x;
                 for (x = 0; x < sdp->sd_journals; x++) {
-                        error = gfs2_recover_journal(gfs2_jdesc_find(sdp, x));
+                        error = gfs2_recover_journal(gfs2_jdesc_find(sdp, x),
+                                                     true);
                         if (error) {
                                 fs_err(sdp, "error recovering journal %u: %d\n",
                                        x, error);
@@ -792,7 +792,7 @@ static int init_journal(struct gfs2_sbd *sdp, int undo)
 
                 gfs2_others_may_mount(sdp);
         } else if (!sdp->sd_args.ar_spectator) {
-                error = gfs2_recover_journal(sdp->sd_jdesc);
+                error = gfs2_recover_journal(sdp->sd_jdesc, true);
                 if (error) {
                         fs_err(sdp, "error recovering my journal: %d\n", error);
                         goto fail_jinode_gh;
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;
 }
 
diff --git a/fs/gfs2/recovery.h b/fs/gfs2/recovery.h
index 1616ac22569a..2226136c7647 100644
--- a/fs/gfs2/recovery.h
+++ b/fs/gfs2/recovery.h
@@ -12,6 +12,8 @@
 
 #include "incore.h"
 
+extern struct workqueue_struct *gfs_recovery_wq;
+
 static inline void gfs2_replay_incr_blk(struct gfs2_sbd *sdp, unsigned int *blk)
 {
         if (++*blk == sdp->sd_jdesc->jd_blocks)
@@ -27,8 +29,8 @@ extern void gfs2_revoke_clean(struct gfs2_sbd *sdp);
 
 extern int gfs2_find_jhead(struct gfs2_jdesc *jd,
                     struct gfs2_log_header_host *head);
-extern int gfs2_recover_journal(struct gfs2_jdesc *gfs2_jd);
-extern struct slow_work_ops gfs2_recover_ops;
+extern int gfs2_recover_journal(struct gfs2_jdesc *gfs2_jd, bool wait);
+extern void gfs2_recover_func(struct work_struct *work);
 
 #endif /* __RECOVERY_DOT_H__ */
 
diff --git a/fs/gfs2/sys.c b/fs/gfs2/sys.c
index d019d0d55e00..ccacffd2faaa 100644
--- a/fs/gfs2/sys.c
+++ b/fs/gfs2/sys.c
@@ -25,6 +25,7 @@
 #include "quota.h"
 #include "util.h"
 #include "glops.h"
+#include "recovery.h"
 
 struct gfs2_attr {
         struct attribute attr;
@@ -376,7 +377,7 @@ static ssize_t recover_store(struct gfs2_sbd *sdp, const char *buf, size_t len)
         list_for_each_entry(jd, &sdp->sd_jindex_list, jd_list) {
                 if (jd->jd_jid != jid)
                         continue;
-                rv = slow_work_enqueue(&jd->jd_work);
+                rv = gfs2_recover_journal(jd, false);
                 break;
         }
 out: