Repository: hibiken/asynq (go)

Issue:
[FEATURE REQUEST] Pause Queue
**Is your feature request related to a problem? Please describe.**
It'd be nice if we can pause a queue without shutting down the background worker process.
Sidekiq Pro has this exact feature documented [here](https://github.com/mperham/sidekiq/wiki/Pro-API#pausing-queues).
If a queue is paused then the background workers shouldn't process any tasks from that queue.

**Describe the solution you'd like**
Add a command to the CLI to pause/unpause a queue.

```sh
asynq queue pause [queue name]

asynq queue unpause [queue name]
```

**Describe alternatives you've considered**
None for now.

**Additional context**
None for now.


Pull request:
Add Feature to Pause Queue
Closes #125 

Patch:
diff --git a/.travis.yml b/.travis.yml
index 9d47394..1d76634 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,6 +5,7 @@ git:
 go: [1.13.x, 1.14.x]
 script:
   - go test -race -v -coverprofile=coverage.txt -covermode=atomic ./...
+  - go test -run=XXX -bench=. -loglevel=debug ./...
 services:
   - redis-server
 after_success:
diff --git a/.travis/benchcmp.sh b/.travis/benchcmp.sh
index b66801c..374efb4 100755
--- a/.travis/benchcmp.sh
+++ b/.travis/benchcmp.sh
@@ -3,13 +3,16 @@ if [ "${TRAVIS_PULL_REQUEST_BRANCH:-$TRAVIS_BRANCH}" != "master" ]; then
     cd ${TRAVIS_BUILD_DIR}/.. && \
     git clone ${REMOTE_URL} "${TRAVIS_REPO_SLUG}-bench" && \
     cd "${TRAVIS_REPO_SLUG}-bench" && \
+
     # Benchmark master
     git checkout master && \
     go test -run=XXX -bench=. ./... > master.txt && \
+
     # Benchmark feature branch
     git checkout ${TRAVIS_COMMIT} && \
     go test -run=XXX -bench=. ./... > feature.txt && \
-    go get -u golang.org/x/tools/cmd/benchcmp && \
+
     # compare two benchmarks
+    go get -u golang.org/x/tools/cmd/benchcmp && \
     benchcmp master.txt feature.txt;
-fi
\ No newline at end of file
+fi
diff --git a/internal/base/base.go b/internal/base/base.go
index e6bd364..2210410 100644
--- a/internal/base/base.go
+++ b/internal/base/base.go
@@ -34,6 +34,7 @@ const (
 	RetryQueue      = "asynq:retry"                  // ZSET
 	DeadQueue       = "asynq:dead"                   // ZSET
 	InProgressQueue = "asynq:in_progress"            // LIST
+	PausedQueues    = "asynq:paused"                 // SET
 	CancelChannel   = "asynq:cancel"                 // PubSub channel
 )
 
@@ -250,7 +251,7 @@ type Broker interface {
 	Retry(msg *TaskMessage, processAt time.Time, errMsg string) error
 	Kill(msg *TaskMessage, errMsg string) error
 	RequeueAll() (int64, error)
-	CheckAndEnqueue(qnames ...string) error
+	CheckAndEnqueue() error
 	WriteServerState(info *ServerInfo, workers []*WorkerInfo, ttl time.Duration) error
 	ClearServerState(host string, pid int, serverID string) error
 	CancelationPubSub() (*redis.PubSub, error) // TODO: Need to decouple from redis to support other brokers
diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go
index 79c1c3e..0ba139a 100644
--- a/internal/rdb/inspect.go
+++ b/internal/rdb/inspect.go
@@ -7,6 +7,7 @@ package rdb
 import (
 	"encoding/json"
 	"fmt"
+	"sort"
 	"strings"
 	"time"
 
@@ -25,10 +26,24 @@ type Stats struct {
 	Dead       int
 	Processed  int
 	Failed     int
-	Queues     map[string]int // map of queue name to number of tasks in the queue (e.g., "default": 100, "critical": 20)
+	Queues     []*Queue
 	Timestamp  time.Time
 }
 
+// Queue represents a task queue.
+type Queue struct {
+	// Name of the queue (e.g. "default", "critical").
+	// Note: It doesn't include the prefix "asynq:queues:".
+	Name string
+
+	// Paused indicates whether the queue is paused.
+	// If true, tasks in the queue should not be processed.
+	Paused bool
+
+	// Size is the number of tasks in the queue.
+	Size int
+}
+
 // DailyStats holds aggregate data for a given day.
 type DailyStats struct {
 	Processed int
@@ -143,8 +158,12 @@ func (r *RDB) CurrentStats() (*Stats, error) {
 	if err != nil {
 		return nil, err
 	}
+	paused, err := r.client.SMembersMap(base.PausedQueues).Result()
+	if err != nil {
+		return nil, err
+	}
 	stats := &Stats{
-		Queues:    make(map[string]int),
+		Queues:    make([]*Queue, 0),
 		Timestamp: now,
 	}
 	for i := 0; i < len(data); i += 2 {
@@ -154,7 +173,14 @@ func (r *RDB) CurrentStats() (*Stats, error) {
 		switch {
 		case strings.HasPrefix(key, base.QueuePrefix):
 			stats.Enqueued += val
-			stats.Queues[strings.TrimPrefix(key, base.QueuePrefix)] = val
+			q := Queue{
+				Name: strings.TrimPrefix(key, base.QueuePrefix),
+				Size: val,
+			}
+			if _, exist := paused[key]; exist {
+				q.Paused = true
+			}
+			stats.Queues = append(stats.Queues, &q)
 		case key == base.InProgressQueue:
 			stats.InProgress = val
 		case key == base.ScheduledQueue:
@@ -169,6 +195,9 @@ func (r *RDB) CurrentStats() (*Stats, error) {
 			stats.Failed = val
 		}
 	}
+	sort.Slice(stats.Queues, func(i, j int) bool {
+		return stats.Queues[i].Name < stats.Queues[j].Name
+	})
 	return stats, nil
 }
 
@@ -830,3 +859,33 @@ func (r *RDB) ListWorkers() ([]*base.WorkerInfo, error) {
 	}
 	return workers, nil
 }
+
+// KEYS[1] -> asynq:paused
+// ARGV[1] -> asynq:queues:<qname> - queue to pause
+var pauseCmd = redis.NewScript(`
+local ismem = redis.call("SISMEMBER", KEYS[1], ARGV[1])
+if ismem == 1 then
+	return redis.error_reply("queue is already paused")
+end
+return redis.call("SADD", KEYS[1], ARGV[1])`)
+
+// Pause pauses processing of tasks from the given queue.
+func (r *RDB) Pause(qname string) error {
+	qkey := base.QueueKey(qname)
+	return pauseCmd.Run(r.client, []string{base.PausedQueues}, qkey).Err()
+}
+
+// KEYS[1] -> asynq:paused
+// ARGV[1] -> asynq:queues:<qname> - queue to unpause
+var unpauseCmd = redis.NewScript(`
+local ismem = redis.call("SISMEMBER", KEYS[1], ARGV[1])
+if ismem == 0 then
+	return redis.error_reply("queue is not paused")
+end
+return redis.call("SREM", KEYS[1], ARGV[1])`)
+
+// Unpause resumes processing of tasks from the given queue.
+func (r *RDB) Unpause(qname string) error {
+	qkey := base.QueueKey(qname)
+	return unpauseCmd.Run(r.client, []string{base.PausedQueues}, qkey).Err()
+}
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index aed147d..e3a493e 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -103,19 +103,14 @@ func (r *RDB) EnqueueUnique(msg *base.TaskMessage, ttl time.Duration) error {
 }
 
 // Dequeue queries given queues in order and pops a task message if there is one and returns it.
+// Dequeue skips a queue if the queue is paused.
 // If all queues are empty, ErrNoProcessableTask error is returned.
 func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
-	var data string
-	var err error
-	if len(qnames) == 1 {
-		data, err = r.dequeueSingle(base.QueueKey(qnames[0]))
-	} else {
-		var keys []string
-		for _, q := range qnames {
-			keys = append(keys, base.QueueKey(q))
-		}
-		data, err = r.dequeue(keys...)
+	var qkeys []interface{}
+	for _, q := range qnames {
+		qkeys = append(qkeys, base.QueueKey(q))
 	}
+	data, err := r.dequeue(qkeys...)
 	if err == redis.Nil {
 		return nil, ErrNoProcessableTask
 	}
@@ -130,29 +125,26 @@ func (r *RDB) Dequeue(qnames ...string) (*base.TaskMessage, error) {
 	return &msg, nil
 }
 
-func (r *RDB) dequeueSingle(queue string) (data string, err error) {
-	// timeout needed to avoid blocking forever
-	return r.client.BRPopLPush(queue, base.InProgressQueue, time.Second).Result()
-}
-
 // KEYS[1] -> asynq:in_progress
+// KEYS[2] -> asynq:paused
 // ARGV    -> List of queues to query in order
+//
+// dequeueCmd checks whether a queue is paused first, before
+// calling RPOPLPUSH to pop a task from the queue.
 var dequeueCmd = redis.NewScript(`
-local res
 for _, qkey in ipairs(ARGV) do
-	res = redis.call("RPOPLPUSH", qkey, KEYS[1])
-	if res then
-		return res
+	if redis.call("SISMEMBER", KEYS[2], qkey) == 0 then
+		local res = redis.call("RPOPLPUSH", qkey, KEYS[1])
+		if res then
+			return res
+		end
 	end
 end
-return res`)
+return nil`)
 
-func (r *RDB) dequeue(queues ...string) (data string, err error) {
-	var args []interface{}
-	for _, qkey := range queues {
-		args = append(args, qkey)
-	}
-	res, err := dequeueCmd.Run(r.client, []string{base.InProgressQueue}, args...).Result()
+func (r *RDB) dequeue(qkeys ...interface{}) (data string, err error) {
+	res, err := dequeueCmd.Run(r.client,
+		[]string{base.InProgressQueue, base.PausedQueues}, qkeys...).Result()
 	if err != nil {
 		return "", err
 	}
@@ -167,7 +159,10 @@ func (r *RDB) dequeue(queues ...string) (data string, err error) {
 // ARGV[3] -> task ID
 // Note: LREM count ZERO means "remove all elements equal to val"
 var doneCmd = redis.NewScript(`
-redis.call("LREM", KEYS[1], 0, ARGV[1]) 
+local x = redis.call("LREM", KEYS[1], 0, ARGV[1]) 
+if x == 0 then
+  return redis.error_reply("NOT FOUND")
+end
 local n = redis.call("INCR", KEYS[2])
 if tonumber(n) == 1 then
 	redis.call("EXPIREAT", KEYS[2], ARGV[2])
@@ -289,7 +284,10 @@ func (r *RDB) ScheduleUnique(msg *base.TaskMessage, processAt time.Time, ttl tim
 // ARGV[3] -> retry_at UNIX timestamp
 // ARGV[4] -> stats expiration timestamp
 var retryCmd = redis.NewScript(`
-redis.call("LREM", KEYS[1], 0, ARGV[1])
+local x = redis.call("LREM", KEYS[1], 0, ARGV[1])
+if x == 0 then
+  return redis.error_reply("NOT FOUND")
+end
 redis.call("ZADD", KEYS[2], ARGV[3], ARGV[2])
 local n = redis.call("INCR", KEYS[3])
 if tonumber(n) == 1 then
@@ -340,7 +338,10 @@ const (
 // ARGV[5] -> max number of tasks in dead queue (e.g., 100)
 // ARGV[6] -> stats expiration timestamp
 var killCmd = redis.NewScript(`
-redis.call("LREM", KEYS[1], 0, ARGV[1])
+local x = redis.call("LREM", KEYS[1], 0, ARGV[1])
+if x == 0 then
+  return redis.error_reply("NOT FOUND")
+end
 redis.call("ZADD", KEYS[2], ARGV[3], ARGV[2])
 redis.call("ZREMRANGEBYSCORE", KEYS[2], "-inf", ARGV[4])
 redis.call("ZREMRANGEBYRANK", KEYS[2], 0, -ARGV[5])
@@ -404,21 +405,17 @@ func (r *RDB) RequeueAll() (int64, error) {
 	return n, nil
 }
 
-// CheckAndEnqueue checks for all scheduled tasks and enqueues any tasks that
-// have to be processed.
-//
-// qnames specifies to which queues to send tasks.
-func (r *RDB) CheckAndEnqueue(qnames ...string) error {
+// CheckAndEnqueue checks for all scheduled/retry tasks and enqueues any tasks that
+// are ready to be processed.
+func (r *RDB) CheckAndEnqueue() (err error) {
 	delayed := []string{base.ScheduledQueue, base.RetryQueue}
 	for _, zset := range delayed {
-		var err error
-		if len(qnames) == 1 {
-			err = r.forwardSingle(zset, base.QueueKey(qnames[0]))
-		} else {
-			err = r.forward(zset)
-		}
-		if err != nil {
-			return err
+		n := 1
+		for n != 0 {
+			n, err = r.forward(zset)
+			if err != nil {
+				return err
+			}
 		}
 	}
 	return nil
@@ -427,40 +424,27 @@ func (r *RDB) CheckAndEnqueue(qnames ...string) error {
 // KEYS[1] -> source queue (e.g. scheduled or retry queue)
 // ARGV[1] -> current unix time
 // ARGV[2] -> queue prefix
+// Note: Script moves tasks up to 100 at a time to keep the runtime of script short.
 var forwardCmd = redis.NewScript(`
-local msgs = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1])
+local msgs = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1], "LIMIT", 0, 100)
 for _, msg in ipairs(msgs) do
 	local decoded = cjson.decode(msg)
 	local qkey = ARGV[2] .. decoded["Queue"]
 	redis.call("LPUSH", qkey, msg)
 	redis.call("ZREM", KEYS[1], msg)
 end
-return msgs`)
-
-// forward moves all tasks with a score less than the current unix time
-// from the src zset.
-func (r *RDB) forward(src string) error {
-	now := float64(time.Now().Unix())
-	return forwardCmd.Run(r.client,
-		[]string{src}, now, base.QueuePrefix).Err()
-}
-
-// KEYS[1] -> source queue (e.g. scheduled or retry queue)
-// KEYS[2] -> destination queue
-var forwardSingleCmd = redis.NewScript(`
-local msgs = redis.call("ZRANGEBYSCORE", KEYS[1], "-inf", ARGV[1])
-for _, msg in ipairs(msgs) do
-	redis.call("LPUSH", KEYS[2], msg)
-	redis.call("ZREM", KEYS[1], msg)
-end
-return msgs`)
+return table.getn(msgs)`)
 
-// forwardSingle moves all tasks with a score less than the current unix time
-// from the src zset to dst list.
-func (r *RDB) forwardSingle(src, dst string) error {
+// forward moves tasks with a score less than the current unix time
+// from the src zset. It returns the number of tasks moved.
+func (r *RDB) forward(src string) (int, error) {
 	now := float64(time.Now().Unix())
-	return forwardSingleCmd.Run(r.client,
-		[]string{src, dst}, now).Err()
+	res, err := forwardCmd.Run(r.client,
+		[]string{src}, now, base.QueuePrefix).Result()
+	if err != nil {
+		return 0, err
+	}
+	return cast.ToInt(res), nil
 }
 
 // KEYS[1]  -> asynq:servers:<host:pid:sid>
diff --git a/processor.go b/processor.go
index 6b080dc..f2bb1af 100644
--- a/processor.go
+++ b/processor.go
@@ -166,14 +166,12 @@ func (p *processor) exec() {
 	msg, err := p.broker.Dequeue(qnames...)
 	switch {
 	case err == rdb.ErrNoProcessableTask:
-		// queues are empty, this is a normal behavior.
-		if len(qnames) > 1 {
-			// sleep to avoid slamming redis and let scheduler move tasks into queues.
-			// Note: With multiple queues, we are not using blocking pop operation and
-			// polling queues instead. This adds significant load to redis.
-			time.Sleep(time.Second)
-		}
 		p.logger.Debug("All queues are empty")
+		// Queues are empty, this is a normal behavior.
+		// Sleep to avoid slamming redis and let scheduler move tasks into queues.
+		// Note: We are not using blocking pop operation and polling queues instead.
+		// This adds significant load to redis.
+		time.Sleep(time.Second)
 		return
 	case err != nil:
 		if p.errLogLimiter.Allow() {
@@ -255,7 +253,7 @@ func (p *processor) requeue(msg *base.TaskMessage) {
 func (p *processor) markAsDone(msg *base.TaskMessage) {
 	err := p.broker.Done(msg)
 	if err != nil {
-		errMsg := fmt.Sprintf("Could not remove task id=%s from %q", msg.ID, base.InProgressQueue)
+		errMsg := fmt.Sprintf("Could not remove task id=%s type=%q from %q", msg.ID, msg.Type, base.InProgressQueue)
 		p.logger.Warnf("%s; Will retry syncing", errMsg)
 		p.syncRequestCh <- &syncRequest{
 			fn: func() error {
diff --git a/scheduler.go b/scheduler.go
index 21dea52..16fd5cc 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -21,29 +21,20 @@ type scheduler struct {
 
 	// poll interval on average
 	avgInterval time.Duration
-
-	// list of queues to move the tasks into.
-	qnames []string
 }
 
 type schedulerParams struct {
 	logger   *log.Logger
 	broker   base.Broker
 	interval time.Duration
-	queues   map[string]int
 }
 
 func newScheduler(params schedulerParams) *scheduler {
-	var qnames []string
-	for q := range params.queues {
-		qnames = append(qnames, q)
-	}
 	return &scheduler{
 		logger:      params.logger,
 		broker:      params.broker,
 		done:        make(chan struct{}),
 		avgInterval: params.interval,
-		qnames:      qnames,
 	}
 }
 
@@ -71,7 +62,7 @@ func (s *scheduler) start(wg *sync.WaitGroup) {
 }
 
 func (s *scheduler) exec() {
-	if err := s.broker.CheckAndEnqueue(s.qnames...); err != nil {
+	if err := s.broker.CheckAndEnqueue(); err != nil {
 		s.logger.Errorf("Could not enqueue scheduled tasks: %v", err)
 	}
 }
diff --git a/server.go b/server.go
index 06ca84a..896c03c 100644
--- a/server.go
+++ b/server.go
@@ -283,8 +283,8 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 	logger.SetLevel(toInternalLogLevel(loglevel))
 
 	rdb := rdb.NewRDB(createRedisClient(r))
-	starting := make(chan *base.TaskMessage, n)
-	finished := make(chan *base.TaskMessage, n)
+	starting := make(chan *base.TaskMessage)
+	finished := make(chan *base.TaskMessage)
 	syncCh := make(chan *syncRequest)
 	status := base.NewServerStatus(base.StatusIdle)
 	cancels := base.NewCancelations()
@@ -309,7 +309,6 @@ func NewServer(r RedisConnOpt, cfg Config) *Server {
 		logger:   logger,
 		broker:   rdb,
 		interval: 5 * time.Second,
-		queues:   queues,
 	})
 	subscriber := newSubscriber(subscriberParams{
 		logger:       logger,
diff --git a/tools/asynq/cmd/pause.go b/tools/asynq/cmd/pause.go
new file mode 100644
index 0000000..091783b
--- /dev/null
+++ b/tools/asynq/cmd/pause.go
@@ -0,0 +1,47 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// pauseCmd represents the pause command
+var pauseCmd = &cobra.Command{
+	Use:   "pause [queue name]",
+	Short: "Pauses the specified queue",
+	Long: `Pause (asynq pause) will pause the specified queue.
+Asynq servers will not process tasks from paused queues.
+Use the "unpause" command to resume a paused queue.
+
+Example: asynq pause default -> Pause the "default" queue`,
+	Args: cobra.ExactValidArgs(1),
+	Run:  pause,
+}
+
+func init() {
+	rootCmd.AddCommand(pauseCmd)
+}
+
+func pause(cmd *cobra.Command, args []string) {
+	c := redis.NewClient(&redis.Options{
+		Addr:     viper.GetString("uri"),
+		DB:       viper.GetInt("db"),
+		Password: viper.GetString("password"),
+	})
+	r := rdb.NewRDB(c)
+	err := r.Pause(args[0])
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Successfully paused queue %q\n", args[0])
+}
diff --git a/tools/asynq/cmd/stats.go b/tools/asynq/cmd/stats.go
index 5ffd026..8db492e 100644
--- a/tools/asynq/cmd/stats.go
+++ b/tools/asynq/cmd/stats.go
@@ -7,7 +7,6 @@ package cmd
 import (
 	"fmt"
 	"os"
-	"sort"
 	"strconv"
 	"strings"
 	"text/tabwriter"
@@ -96,24 +95,31 @@ func printStates(s *rdb.Stats) {
 	tw.Flush()
 }
 
-func printQueues(queues map[string]int) {
-	var qnames, seps, counts []string
-	for q := range queues {
-		qnames = append(qnames, strings.Title(q))
+func printQueues(queues []*rdb.Queue) {
+	var headers, seps, counts []string
+	for _, q := range queues {
+		title := queueTitle(q)
+		headers = append(headers, title)
+		seps = append(seps, strings.Repeat("-", len(title)))
+		counts = append(counts, strconv.Itoa(q.Size))
 	}
-	sort.Strings(qnames) // sort for stable order
-	for _, q := range qnames {
-		seps = append(seps, strings.Repeat("-", len(q)))
-		counts = append(counts, strconv.Itoa(queues[strings.ToLower(q)]))
-	}
-	format := strings.Repeat("%v\t", len(qnames)) + "\n"
+	format := strings.Repeat("%v\t", len(headers)) + "\n"
 	tw := new(tabwriter.Writer).Init(os.Stdout, 0, 8, 2, ' ', 0)
-	fmt.Fprintf(tw, format, toInterfaceSlice(qnames)...)
+	fmt.Fprintf(tw, format, toInterfaceSlice(headers)...)
 	fmt.Fprintf(tw, format, toInterfaceSlice(seps)...)
 	fmt.Fprintf(tw, format, toInterfaceSlice(counts)...)
 	tw.Flush()
 }
 
+func queueTitle(q *rdb.Queue) string {
+	var b strings.Builder
+	b.WriteString(strings.Title(q.Name))
+	if q.Paused {
+		b.WriteString(" (Paused)")
+	}
+	return b.String()
+}
+
 func printStats(s *rdb.Stats) {
 	format := strings.Repeat("%v\t", 3) + "\n"
 	tw := new(tabwriter.Writer).Init(os.Stdout, 0, 8, 2, ' ', 0)
diff --git a/tools/asynq/cmd/unpause.go b/tools/asynq/cmd/unpause.go
new file mode 100644
index 0000000..79e0764
--- /dev/null
+++ b/tools/asynq/cmd/unpause.go
@@ -0,0 +1,46 @@
+// Copyright 2020 Kentaro Hibino. All rights reserved.
+// Use of this source code is governed by a MIT license
+// that can be found in the LICENSE file.
+
+package cmd
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hibiken/asynq/internal/rdb"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+// unpauseCmd represents the unpause command
+var unpauseCmd = &cobra.Command{
+	Use:   "unpause [queue name]",
+	Short: "Unpauses the specified queue",
+	Long: `Unpause (asynq unpause) will unpause the specified queue.
+Asynq servers will process tasks from unpaused/resumed queues.
+
+Example: asynq unpause default -> Resume the "default" queue`,
+	Args: cobra.ExactValidArgs(1),
+	Run:  unpause,
+}
+
+func init() {
+	rootCmd.AddCommand(unpauseCmd)
+}
+
+func unpause(cmd *cobra.Command, args []string) {
+	c := redis.NewClient(&redis.Options{
+		Addr:     viper.GetString("uri"),
+		DB:       viper.GetInt("db"),
+		Password: viper.GetString("password"),
+	})
+	r := rdb.NewRDB(c)
+	err := r.Unpause(args[0])
+	if err != nil {
+		fmt.Printf("error: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Successfully resumed queue %q\n", args[0])
+}


Test patch:
diff --git a/benchmark_test.go b/benchmark_test.go
index 8f4913c..59b9e7e 100644
--- a/benchmark_test.go
+++ b/benchmark_test.go
@@ -7,7 +7,6 @@ package asynq
 import (
 	"context"
 	"fmt"
-	"math/rand"
 	"sync"
 	"testing"
 	"time"
@@ -29,6 +28,7 @@ func BenchmarkEndToEndSimple(b *testing.B) {
 			RetryDelayFunc: func(n int, err error, t *Task) time.Duration {
 				return time.Second
 			},
+			LogLevel: testLogLevel,
 		})
 		// Create a bunch of tasks
 		for i := 0; i < count; i++ {
@@ -60,7 +60,6 @@ func BenchmarkEndToEnd(b *testing.B) {
 	const count = 100000
 	for n := 0; n < b.N; n++ {
 		b.StopTimer() // begin setup
-		rand.Seed(time.Now().UnixNano())
 		setup(b)
 		redis := &RedisClientOpt{
 			Addr: redisAddr,
@@ -72,6 +71,7 @@ func BenchmarkEndToEnd(b *testing.B) {
 			RetryDelayFunc: func(n int, err error, t *Task) time.Duration {
 				return time.Second
 			},
+			LogLevel: testLogLevel,
 		})
 		// Create a bunch of tasks
 		for i := 0; i < count; i++ {
@@ -90,8 +90,16 @@ func BenchmarkEndToEnd(b *testing.B) {
 		var wg sync.WaitGroup
 		wg.Add(count * 2)
 		handler := func(ctx context.Context, t *Task) error {
-			// randomly fail 1% of tasks
-			if rand.Intn(100) == 1 {
+			n, err := t.Payload.GetInt("data")
+			if err != nil {
+				b.Logf("internal error: %v", err)
+			}
+			retried, ok := GetRetryCount(ctx)
+			if !ok {
+				b.Logf("internal error: %v", err)
+			}
+			// Fail 1% of tasks for the first attempt.
+			if retried == 0 && n%100 == 0 {
 				return fmt.Errorf(":(")
 			}
 			wg.Done()
@@ -131,6 +139,7 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 				"default": 3,
 				"low":     1,
 			},
+			LogLevel: testLogLevel,
 		})
 		// Create a bunch of tasks
 		for i := 0; i < highCount; i++ {
@@ -168,3 +177,62 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 		b.StartTimer() // end teardown
 	}
 }
+
+// E2E benchmark to check client enqueue operation performs correctly,
+// while server is busy processing tasks.
+func BenchmarkClientWhileServerRunning(b *testing.B) {
+	const count = 10000
+	for n := 0; n < b.N; n++ {
+		b.StopTimer() // begin setup
+		setup(b)
+		redis := &RedisClientOpt{
+			Addr: redisAddr,
+			DB:   redisDB,
+		}
+		client := NewClient(redis)
+		srv := NewServer(redis, Config{
+			Concurrency: 10,
+			RetryDelayFunc: func(n int, err error, t *Task) time.Duration {
+				return time.Second
+			},
+			LogLevel: testLogLevel,
+		})
+		// Enqueue 10,000 tasks.
+		for i := 0; i < count; i++ {
+			t := NewTask(fmt.Sprintf("task%d", i), map[string]interface{}{"data": i})
+			if err := client.Enqueue(t); err != nil {
+				b.Fatalf("could not enqueue a task: %v", err)
+			}
+		}
+		// Schedule 10,000 tasks.
+		for i := 0; i < count; i++ {
+			t := NewTask(fmt.Sprintf("scheduled%d", i), map[string]interface{}{"data": i})
+			if err := client.EnqueueAt(time.Now().Add(time.Second), t); err != nil {
+				b.Fatalf("could not enqueue a task: %v", err)
+			}
+		}
+
+		handler := func(ctx context.Context, t *Task) error {
+			return nil
+		}
+		srv.Start(HandlerFunc(handler))
+
+		b.StartTimer() // end setup
+
+		b.Log("Starting enqueueing")
+		enqueued := 0
+		for enqueued < 100000 {
+			t := NewTask(fmt.Sprintf("enqueued%d", enqueued), map[string]interface{}{"data": enqueued})
+			if err := client.Enqueue(t); err != nil {
+				b.Logf("could not enqueue task %d: %v", enqueued, err)
+				continue
+			}
+			enqueued++
+		}
+		b.Logf("Finished enqueueing %d tasks", enqueued)
+
+		b.StopTimer() // begin teardown
+		srv.Stop()
+		b.StartTimer() // end teardown
+	}
+}
diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go
index c0b2c09..530df64 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -38,6 +38,7 @@ func TestCurrentStats(t *testing.T) {
 		processed  int
 		failed     int
 		allQueues  []interface{}
+		paused     []string
 		want       *Stats
 	}{
 		{
@@ -55,6 +56,7 @@ func TestCurrentStats(t *testing.T) {
 			processed: 120,
 			failed:    2,
 			allQueues: []interface{}{base.DefaultQueue, base.QueueKey("critical"), base.QueueKey("low")},
+			paused:    []string{},
 			want: &Stats{
 				Enqueued:   3,
 				InProgress: 1,
@@ -64,7 +66,12 @@ func TestCurrentStats(t *testing.T) {
 				Processed:  120,
 				Failed:     2,
 				Timestamp:  now,
-				Queues:     map[string]int{base.DefaultQueueName: 1, "critical": 1, "low": 1},
+				// Queues should be sorted by name.
+				Queues: []*Queue{
+					{Name: "critical", Paused: false, Size: 1},
+					{Name: "default", Paused: false, Size: 1},
+					{Name: "low", Paused: false, Size: 1},
+				},
 			},
 		},
 		{
@@ -82,6 +89,7 @@ func TestCurrentStats(t *testing.T) {
 			processed: 90,
 			failed:    10,
 			allQueues: []interface{}{base.DefaultQueue},
+			paused:    []string{},
 			want: &Stats{
 				Enqueued:   0,
 				InProgress: 0,
@@ -91,13 +99,52 @@ func TestCurrentStats(t *testing.T) {
 				Processed:  90,
 				Failed:     10,
 				Timestamp:  now,
-				Queues:     map[string]int{base.DefaultQueueName: 0},
+				Queues: []*Queue{
+					{Name: base.DefaultQueueName, Paused: false, Size: 0},
+				},
+			},
+		},
+		{
+			enqueued: map[string][]*base.TaskMessage{
+				base.DefaultQueueName: {m1},
+				"critical":            {m5},
+				"low":                 {m6},
+			},
+			inProgress: []*base.TaskMessage{m2},
+			scheduled: []h.ZSetEntry{
+				{Msg: m3, Score: float64(now.Add(time.Hour).Unix())},
+				{Msg: m4, Score: float64(now.Unix())}},
+			retry:     []h.ZSetEntry{},
+			dead:      []h.ZSetEntry{},
+			processed: 120,
+			failed:    2,
+			allQueues: []interface{}{base.DefaultQueue, base.QueueKey("critical"), base.QueueKey("low")},
+			paused:    []string{"critical", "low"},
+			want: &Stats{
+				Enqueued:   3,
+				InProgress: 1,
+				Scheduled:  2,
+				Retry:      0,
+				Dead:       0,
+				Processed:  120,
+				Failed:     2,
+				Timestamp:  now,
+				Queues: []*Queue{
+					{Name: "critical", Paused: true, Size: 1},
+					{Name: "default", Paused: false, Size: 1},
+					{Name: "low", Paused: true, Size: 1},
+				},
 			},
 		},
 	}
 
 	for _, tc := range tests {
 		h.FlushDB(t, r.client) // clean up db before each test case
+		for _, qname := range tc.paused {
+			if err := r.Pause(qname); err != nil {
+				t.Fatal(err)
+			}
+		}
 		for qname, msgs := range tc.enqueued {
 			h.SeedEnqueuedQueue(t, r.client, msgs, qname)
 		}
@@ -136,7 +183,7 @@ func TestCurrentStatsWithoutData(t *testing.T) {
 		Processed:  0,
 		Failed:     0,
 		Timestamp:  time.Now(),
-		Queues:     map[string]int{},
+		Queues:     make([]*Queue, 0),
 	}
 
 	got, err := r.CurrentStats()
@@ -658,12 +705,14 @@ func TestListRetry(t *testing.T) {
 func TestListRetryPagination(t *testing.T) {
 	r := setup(t)
 	// create 100 tasks with an increasing number of wait time.
+	now := time.Now()
+	var seed []h.ZSetEntry
 	for i := 0; i < 100; i++ {
 		msg := h.NewTaskMessage(fmt.Sprintf("task %d", i), nil)
-		if err := r.Retry(msg, time.Now().Add(time.Duration(i)*time.Second), "error"); err != nil {
-			t.Fatal(err)
-		}
+		processAt := now.Add(time.Duration(i) * time.Second)
+		seed = append(seed, h.ZSetEntry{Msg: msg, Score: float64(processAt.Unix())})
 	}
+	h.SeedRetryQueue(t, r.client, seed)
 
 	tests := []struct {
 		desc      string
@@ -2156,3 +2205,160 @@ func TestListWorkers(t *testing.T) {
 		}
 	}
 }
+
+func TestPause(t *testing.T) {
+	r := setup(t)
+
+	tests := []struct {
+		initial []string // initial keys in the paused set
+		qname   string   // name of the queue to pause
+		want    []string // expected keys in the paused set
+	}{
+		{[]string{}, "default", []string{"asynq:queues:default"}},
+		{[]string{"asynq:queues:default"}, "critical", []string{"asynq:queues:default", "asynq:queues:critical"}},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+
+		// Set up initial state.
+		for _, qkey := range tc.initial {
+			if err := r.client.SAdd(base.PausedQueues, qkey).Err(); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Pause(tc.qname)
+		if err != nil {
+			t.Errorf("Pause(%q) returned error: %v", tc.qname, err)
+		}
+
+		got, err := r.client.SMembers(base.PausedQueues).Result()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortStringSliceOpt); diff != "" {
+			t.Errorf("%q has members %v, want %v; (-want,+got)\n%s",
+				base.PausedQueues, got, tc.want, diff)
+		}
+	}
+}
+
+func TestPauseError(t *testing.T) {
+	r := setup(t)
+
+	tests := []struct {
+		desc    string   // test case description
+		initial []string // initial keys in the paused set
+		qname   string   // name of the queue to pause
+		want    []string // expected keys in the paused set
+	}{
+		{"queue already paused", []string{"asynq:queues:default"}, "default", []string{"asynq:queues:default"}},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+
+		// Set up initial state.
+		for _, qkey := range tc.initial {
+			if err := r.client.SAdd(base.PausedQueues, qkey).Err(); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Pause(tc.qname)
+		if err == nil {
+			t.Errorf("%s; Pause(%q) returned nil: want error", tc.desc, tc.qname)
+		}
+
+		got, err := r.client.SMembers(base.PausedQueues).Result()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortStringSliceOpt); diff != "" {
+			t.Errorf("%s; %q has members %v, want %v; (-want,+got)\n%s",
+				tc.desc, base.PausedQueues, got, tc.want, diff)
+		}
+	}
+}
+
+func TestUnpause(t *testing.T) {
+	r := setup(t)
+
+	tests := []struct {
+		initial []string // initial keys in the paused set
+		qname   string   // name of the queue to unpause
+		want    []string // expected keys in the paused set
+	}{
+		{[]string{"asynq:queues:default"}, "default", []string{}},
+		{[]string{"asynq:queues:default", "asynq:queues:low"}, "low", []string{"asynq:queues:default"}},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+
+		// Set up initial state.
+		for _, qkey := range tc.initial {
+			if err := r.client.SAdd(base.PausedQueues, qkey).Err(); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Unpause(tc.qname)
+		if err != nil {
+			t.Errorf("Unpause(%q) returned error: %v", tc.qname, err)
+		}
+
+		got, err := r.client.SMembers(base.PausedQueues).Result()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortStringSliceOpt); diff != "" {
+			t.Errorf("%q has members %v, want %v; (-want,+got)\n%s",
+				base.PausedQueues, got, tc.want, diff)
+		}
+	}
+}
+
+func TestUnpauseError(t *testing.T) {
+	r := setup(t)
+
+	tests := []struct {
+		desc    string   // test case description
+		initial []string // initial keys in the paused set
+		qname   string   // name of the queue to unpause
+		want    []string // expected keys in the paused set
+	}{
+		{"set is empty", []string{}, "default", []string{}},
+		{"queue is not in the set", []string{"asynq:queues:default"}, "low", []string{"asynq:queues:default"}},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client)
+
+		// Set up initial state.
+		for _, qkey := range tc.initial {
+			if err := r.client.SAdd(base.PausedQueues, qkey).Err(); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		err := r.Unpause(tc.qname)
+		if err == nil {
+			t.Errorf("%s; Unpause(%q) returned nil: want error", tc.desc, tc.qname)
+		}
+
+		got, err := r.client.SMembers(base.PausedQueues).Result()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if diff := cmp.Diff(tc.want, got, h.SortStringSliceOpt); diff != "" {
+			t.Errorf("%s; %q has members %v, want %v; (-want,+got)\n%s",
+				tc.desc, base.PausedQueues, got, tc.want, diff)
+		}
+	}
+}
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index 2172214..417dd17 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -227,6 +227,97 @@ func TestDequeue(t *testing.T) {
 	}
 }
 
+func TestDequeueIgnoresPausedQueues(t *testing.T) {
+	r := setup(t)
+	t1 := h.NewTaskMessage("send_email", map[string]interface{}{"subject": "hello!"})
+	t2 := h.NewTaskMessage("export_csv", nil)
+
+	tests := []struct {
+		paused         []string // list of paused queues
+		enqueued       map[string][]*base.TaskMessage
+		args           []string // list of queues to query
+		want           *base.TaskMessage
+		err            error
+		wantEnqueued   map[string][]*base.TaskMessage
+		wantInProgress []*base.TaskMessage
+	}{
+		{
+			paused: []string{"default"},
+			enqueued: map[string][]*base.TaskMessage{
+				"default":  {t1},
+				"critical": {t2},
+			},
+			args: []string{"default", "critical"},
+			want: t2,
+			err:  nil,
+			wantEnqueued: map[string][]*base.TaskMessage{
+				"default":  {t1},
+				"critical": {},
+			},
+			wantInProgress: []*base.TaskMessage{t2},
+		},
+		{
+			paused: []string{"default"},
+			enqueued: map[string][]*base.TaskMessage{
+				"default": {t1},
+			},
+			args: []string{"default"},
+			want: nil,
+			err:  ErrNoProcessableTask,
+			wantEnqueued: map[string][]*base.TaskMessage{
+				"default": {t1},
+			},
+			wantInProgress: []*base.TaskMessage{},
+		},
+		{
+			paused: []string{"critical", "default"},
+			enqueued: map[string][]*base.TaskMessage{
+				"default":  {t1},
+				"critical": {t2},
+			},
+			args: []string{"default", "critical"},
+			want: nil,
+			err:  ErrNoProcessableTask,
+			wantEnqueued: map[string][]*base.TaskMessage{
+				"default":  {t1},
+				"critical": {t2},
+			},
+			wantInProgress: []*base.TaskMessage{},
+		},
+	}
+
+	for _, tc := range tests {
+		h.FlushDB(t, r.client) // clean up db before each test case
+		for _, qname := range tc.paused {
+			if err := r.Pause(qname); err != nil {
+				t.Fatal(err)
+			}
+		}
+		for queue, msgs := range tc.enqueued {
+			h.SeedEnqueuedQueue(t, r.client, msgs, queue)
+		}
+
+		got, err := r.Dequeue(tc.args...)
+		if !cmp.Equal(got, tc.want) || err != tc.err {
+			t.Errorf("Dequeue(%v) = %v, %v; want %v, %v",
+				tc.args, got, err, tc.want, tc.err)
+			continue
+		}
+
+		for queue, want := range tc.wantEnqueued {
+			gotEnqueued := h.GetEnqueuedMessages(t, r.client, queue)
+			if diff := cmp.Diff(want, gotEnqueued, h.SortMsgOpt); diff != "" {
+				t.Errorf("mismatch found in %q: (-want,+got):\n%s", base.QueueKey(queue), diff)
+			}
+		}
+
+		gotInProgress := h.GetInProgressMessages(t, r.client)
+		if diff := cmp.Diff(tc.wantInProgress, gotInProgress, h.SortMsgOpt); diff != "" {
+			t.Errorf("mismatch found in %q: (-want,+got):\n%s", base.InProgressQueue, diff)
+		}
+	}
+}
+
 func TestDone(t *testing.T) {
 	r := setup(t)
 	t1 := h.NewTaskMessage("send_email", nil)
@@ -769,7 +860,6 @@ func TestCheckAndEnqueue(t *testing.T) {
 	tests := []struct {
 		scheduled     []h.ZSetEntry
 		retry         []h.ZSetEntry
-		qnames        []string
 		wantEnqueued  map[string][]*base.TaskMessage
 		wantScheduled []*base.TaskMessage
 		wantRetry     []*base.TaskMessage
@@ -781,7 +871,6 @@ func TestCheckAndEnqueue(t *testing.T) {
 			},
 			retry: []h.ZSetEntry{
 				{Msg: t3, Score: float64(secondAgo.Unix())}},
-			qnames: []string{"default"},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {t1, t2, t3},
 			},
@@ -794,7 +883,6 @@ func TestCheckAndEnqueue(t *testing.T) {
 				{Msg: t2, Score: float64(secondAgo.Unix())}},
 			retry: []h.ZSetEntry{
 				{Msg: t3, Score: float64(secondAgo.Unix())}},
-			qnames: []string{"default"},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {t2, t3},
 			},
@@ -807,7 +895,6 @@ func TestCheckAndEnqueue(t *testing.T) {
 				{Msg: t2, Score: float64(hourFromNow.Unix())}},
 			retry: []h.ZSetEntry{
 				{Msg: t3, Score: float64(hourFromNow.Unix())}},
-			qnames: []string{"default"},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default": {},
 			},
@@ -821,7 +908,6 @@ func TestCheckAndEnqueue(t *testing.T) {
 			},
 			retry: []h.ZSetEntry{
 				{Msg: t5, Score: float64(secondAgo.Unix())}},
-			qnames: []string{"default", "critical", "low"},
 			wantEnqueued: map[string][]*base.TaskMessage{
 				"default":  {t1},
 				"critical": {t4},
@@ -837,7 +923,7 @@ func TestCheckAndEnqueue(t *testing.T) {
 		h.SeedScheduledQueue(t, r.client, tc.scheduled)
 		h.SeedRetryQueue(t, r.client, tc.retry)
 
-		err := r.CheckAndEnqueue(tc.qnames...)
+		err := r.CheckAndEnqueue()
 		if err != nil {
 			t.Errorf("(*RDB).CheckScheduled() = %v, want nil", err)
 			continue
diff --git a/internal/testbroker/testbroker.go b/internal/testbroker/testbroker.go
index 8bcba61..9044e17 100644
--- a/internal/testbroker/testbroker.go
+++ b/internal/testbroker/testbroker.go
@@ -132,7 +132,7 @@ func (tb *TestBroker) RequeueAll() (int64, error) {
 	return tb.real.RequeueAll()
 }
 
-func (tb *TestBroker) CheckAndEnqueue(qnames ...string) error {
+func (tb *TestBroker) CheckAndEnqueue() error {
 	tb.mu.Lock()
 	defer tb.mu.Unlock()
 	if tb.sleeping {
diff --git a/processor_test.go b/processor_test.go
index 9ecb8fe..da1b577 100644
--- a/processor_test.go
+++ b/processor_test.go
@@ -104,12 +104,14 @@ func TestProcessorSuccess(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-		time.Sleep(time.Second) // wait for one second to allow all enqueued tasks to be processed.
+		time.Sleep(2 * time.Second) // wait for two second to allow all enqueued tasks to be processed.
 		p.terminate()
 
+		mu.Lock()
 		if diff := cmp.Diff(tc.wantProcessed, processed, sortTaskOpt, cmp.AllowUnexported(Payload{})); diff != "" {
 			t.Errorf("mismatch found in processed tasks; (-want, +got)\n%s", diff)
 		}
+		mu.Unlock()
 
 		if l := r.LLen(base.InProgressQueue).Val(); l != 0 {
 			t.Errorf("%q has %d tasks, want 0", base.InProgressQueue, l)
@@ -160,7 +162,7 @@ func TestProcessorRetry(t *testing.T) {
 			handler: HandlerFunc(func(ctx context.Context, task *Task) error {
 				return fmt.Errorf(errMsg)
 			}),
-			wait: time.Second,
+			wait: 2 * time.Second,
 			wantRetry: []h.ZSetEntry{
 				{Msg: &r2, Score: float64(now.Add(time.Minute).Unix())},
 				{Msg: &r3, Score: float64(now.Add(time.Minute).Unix())},
@@ -217,7 +219,7 @@ func TestProcessorRetry(t *testing.T) {
 				t.Fatal(err)
 			}
 		}
-		time.Sleep(tc.wait)
+		time.Sleep(tc.wait) // FIXME: This makes test flaky.
 		p.terminate()
 
 		cmpOpt := cmpopts.EquateApprox(0, float64(time.Second)) // allow up to a second difference in zset score
diff --git a/scheduler_test.go b/scheduler_test.go
index ee82250..34cd9fb 100644
--- a/scheduler_test.go
+++ b/scheduler_test.go
@@ -23,7 +23,6 @@ func TestScheduler(t *testing.T) {
 		logger:   testLogger,
 		broker:   rdbClient,
 		interval: pollInterval,
-		queues:   defaultQueueConfig,
 	})
 	t1 := h.NewTaskMessage("gen_thumbnail", nil)
 	t2 := h.NewTaskMessage("send_email", nil)