Repository: hibiken/asynq (go)

Issue:
[FEATURE REQUEST] Make handler context aware
We should support timeout per task and allow user to cancel task processing using the CLI.

Cancellation would be useful in two cases:

- User manually cancels a in-progress task using the CLI
- Background worker process is shutting down, so in-progress tasks should be cancelled

Optional timeout per task is used in cases such as:

- For tasks that are best-effort. If processing could not complete within a certain timeout, bail and retry later.

In order to support both cancel and timeout, we need to change the `Handler` interface.

```go
type Handler interface {
    func ProcessTask(ctx context.Context, t *Task) error
}
```

and add `Timeout` function to create a timeout option when scheduling a task.

```go
client.Schedule(task, time.Now().Add(time.Hour), asynq.Timeout(10 * time.Second))
```

CLI will have a new command to cancel a in-progress task:

```sh
asynqmon cancel [task_id]
```

A cancelled task should be moved to *retry* state consuming a retry count.

Pull request:
Add support for cancelation and timeout
Closes #69 

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7de289e..b9c73e1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed
+
+- `Handler` interface has changed. `ProcessTask` method takes two arguments `context.Context` and `*asynq.Task`
+- `Queues` field in `Config` has change from `map[string]uint` to `map[string]int`
+
+### Added
+
+- `Client` can optionally schedule task with `asynq.Timeout(duration)` to specify timeout duration for task. Default is no timeout.
+- `asynqmon cancel [task id]` will send a cancelation signal to the goroutine processing the speicified task.
+
 ## [0.3.0] - 2020-02-04
 
 ### Added
diff --git a/README.md b/README.md
index 1e80312..ec9791c 100644
--- a/README.md
+++ b/README.md
@@ -11,8 +11,6 @@ It is backed by Redis and it is designed to have a low barrier to entry. It shou
 
 **Important Note**: Current major version is zero (v0.x.x) to accomodate rapid development and fast iteration while getting early feedback from users. The public API could change without a major version update before v1.0.0 release.
 
-[Quick Tour of Asynq (YouTube video)](https://www.youtube.com/watch?v=FzMJdddNcmA&t=62s)
-
 ![Gif](/docs/assets/demo.gif)
 
 ## Installation
@@ -57,6 +55,9 @@ func main() {
 
     // Use custom queue called "critical"
     err = client.Schedule(t1, time.Now(), asynq.Queue("critical"))
+
+    // Use timeout to specify how long a task may run (Default is no limit)
+    err = client.Schedule(t1, time.Now(), asynq.Timeout(30 * time.Second))
 }
 ```
 
@@ -71,9 +72,8 @@ func main() {
     bg := asynq.NewBackground(r, &asynq.Config{
         // Specify how many concurrent workers to use
         Concurrency: 10,
-        // You can optionally create multiple queues
-        // with different priority level
-        Queues: map[string]uint{
+        // You can optionally create multiple queues with different priority.
+        Queues: map[string]int{
             "critical": 6,
             "default":  3,
             "low":      1,
@@ -94,7 +94,7 @@ func main() {
 // If ProcessTask return a non-nil error or panics, the task
 // will be retried after delay.
 type Handler interface {
-    ProcessTask(*Task) error
+    ProcessTask(context.Context, *asynq.Task) error
 }
 ```
 
diff --git a/background.go b/background.go
index 273be7d..fa533a2 100644
--- a/background.go
+++ b/background.go
@@ -5,6 +5,7 @@
 package asynq
 
 import (
+	"context"
 	"fmt"
 	"math"
 	"math/rand"
@@ -39,13 +40,14 @@ type Background struct {
 	processor   *processor
 	syncer      *syncer
 	heartbeater *heartbeater
+	subscriber  *subscriber
 }
 
 // Config specifies the background-task processing behavior.
 type Config struct {
 	// Maximum number of concurrent processing of tasks.
 	//
-	// If set to zero or negative value, NewBackground will overwrite the value to one.
+	// If set to a zero or negative value, NewBackground will overwrite the value to one.
 	Concurrency int
 
 	// Function to calculate retry delay for a failed task.
@@ -57,15 +59,15 @@ type Config struct {
 	// t is the task in question.
 	RetryDelayFunc func(n int, e error, t *Task) time.Duration
 
-	// List of queues to process with given priority level. Keys are the names of the
-	// queues and values are associated priority level.
+	// List of queues to process with given priority value. Keys are the names of the
+	// queues and values are associated priority value.
 	//
 	// If set to nil or not specified, the background will process only the "default" queue.
 	//
 	// Priority is treated as follows to avoid starving low priority queues.
 	//
 	// Example:
-	// Queues: map[string]uint{
+	// Queues: map[string]int{
 	//     "critical": 6,
 	//     "default":  3,
 	//     "low":      1,
@@ -73,7 +75,9 @@ type Config struct {
 	// With the above config and given that all queues are not empty, the tasks
 	// in "critical", "default", "low" should be processed 60%, 30%, 10% of
 	// the time respectively.
-	Queues map[string]uint
+	//
+	// If a queue has a zero or negative priority value, the queue will be ignored.
+	Queues map[string]int
 
 	// StrictPriority indicates whether the queue priority should be treated strictly.
 	//
@@ -90,7 +94,7 @@ func defaultDelayFunc(n int, e error, t *Task) time.Duration {
 	return time.Duration(s) * time.Second
 }
 
-var defaultQueueConfig = map[string]uint{
+var defaultQueueConfig = map[string]int{
 	base.DefaultQueueName: 1,
 }
 
@@ -105,8 +109,13 @@ func NewBackground(r RedisConnOpt, cfg *Config) *Background {
 	if delayFunc == nil {
 		delayFunc = defaultDelayFunc
 	}
-	queues := cfg.Queues
-	if queues == nil || len(queues) == 0 {
+	queues := make(map[string]int)
+	for qname, p := range cfg.Queues {
+		if p > 0 {
+			queues[qname] = p
+		}
+	}
+	if len(queues) == 0 {
 		queues = defaultQueueConfig
 	}
 
@@ -119,10 +128,12 @@ func NewBackground(r RedisConnOpt, cfg *Config) *Background {
 	pinfo := base.NewProcessInfo(host, pid, n, queues, cfg.StrictPriority)
 	rdb := rdb.NewRDB(createRedisClient(r))
 	syncRequestCh := make(chan *syncRequest)
+	cancelations := base.NewCancelations()
 	syncer := newSyncer(syncRequestCh, 5*time.Second)
 	heartbeater := newHeartbeater(rdb, pinfo, 5*time.Second)
 	scheduler := newScheduler(rdb, 5*time.Second, queues)
-	processor := newProcessor(rdb, pinfo, delayFunc, syncRequestCh)
+	processor := newProcessor(rdb, pinfo, delayFunc, syncRequestCh, cancelations)
+	subscriber := newSubscriber(rdb, cancelations)
 	return &Background{
 		pinfo:       pinfo,
 		rdb:         rdb,
@@ -130,6 +141,7 @@ func NewBackground(r RedisConnOpt, cfg *Config) *Background {
 		processor:   processor,
 		syncer:      syncer,
 		heartbeater: heartbeater,
+		subscriber:  subscriber,
 	}
 }
 
@@ -141,18 +153,18 @@ func NewBackground(r RedisConnOpt, cfg *Config) *Background {
 // If ProcessTask return a non-nil error or panics, the task
 // will be retried after delay.
 type Handler interface {
-	ProcessTask(*Task) error
+	ProcessTask(context.Context, *Task) error
 }
 
 // The HandlerFunc type is an adapter to allow the use of
 // ordinary functions as a Handler. If f is a function
 // with the appropriate signature, HandlerFunc(f) is a
 // Handler that calls f.
-type HandlerFunc func(*Task) error
+type HandlerFunc func(context.Context, *Task) error
 
-// ProcessTask calls fn(task)
-func (fn HandlerFunc) ProcessTask(task *Task) error {
-	return fn(task)
+// ProcessTask calls fn(ctx, task)
+func (fn HandlerFunc) ProcessTask(ctx context.Context, task *Task) error {
+	return fn(ctx, task)
 }
 
 // Run starts the background-task processing and blocks until
@@ -197,6 +209,7 @@ func (bg *Background) start(handler Handler) {
 	bg.processor.handler = handler
 
 	bg.heartbeater.start()
+	bg.subscriber.start()
 	bg.syncer.start()
 	bg.scheduler.start()
 	bg.processor.start()
@@ -215,6 +228,7 @@ func (bg *Background) stop() {
 	// Note: processor and all worker goroutines need to be exited
 	// before shutting down syncer to avoid goroutine leak.
 	bg.syncer.terminate()
+	bg.subscriber.terminate()
 	bg.heartbeater.terminate()
 
 	bg.rdb.ClearProcessInfo(bg.pinfo)
diff --git a/client.go b/client.go
index afd1f9e..b7116e4 100644
--- a/client.go
+++ b/client.go
@@ -34,8 +34,9 @@ type Option interface{}
 
 // Internal option representations.
 type (
-	retryOption int
-	queueOption string
+	retryOption   int
+	queueOption   string
+	timeoutOption time.Duration
 )
 
 // MaxRetry returns an option to specify the max number of times
@@ -56,15 +57,24 @@ func Queue(name string) Option {
 	return queueOption(strings.ToLower(name))
 }
 
+// Timeout returns an option to specify how long a task may run.
+//
+// Zero duration means no limit.
+func Timeout(d time.Duration) Option {
+	return timeoutOption(d)
+}
+
 type option struct {
-	retry int
-	queue string
+	retry   int
+	queue   string
+	timeout time.Duration
 }
 
 func composeOptions(opts ...Option) option {
 	res := option{
-		retry: defaultMaxRetry,
-		queue: base.DefaultQueueName,
+		retry:   defaultMaxRetry,
+		queue:   base.DefaultQueueName,
+		timeout: 0,
 	}
 	for _, opt := range opts {
 		switch opt := opt.(type) {
@@ -72,6 +82,8 @@ func composeOptions(opts ...Option) option {
 			res.retry = int(opt)
 		case queueOption:
 			res.queue = string(opt)
+		case timeoutOption:
+			res.timeout = time.Duration(opt)
 		default:
 			// ignore unexpected option
 		}
@@ -99,6 +111,7 @@ func (c *Client) Schedule(task *Task, processAt time.Time, opts ...Option) error
 		Payload: task.Payload.data,
 		Queue:   opt.queue,
 		Retry:   opt.retry,
+		Timeout: opt.timeout.String(),
 	}
 	return c.enqueue(msg, processAt)
 }
diff --git a/doc.go b/doc.go
index f56b5c4..a369fc1 100644
--- a/doc.go
+++ b/doc.go
@@ -45,7 +45,7 @@ Example of a type that implements the Handler interface.
         // ...
     }
 
-    func (h *TaskHandler) ProcessTask(task *asynq.Task) error {
+    func (h *TaskHandler) ProcessTask(ctx context.Context, task *asynq.Task) error {
         switch task.Type {
         case "send_email":
             id, err := task.Payload.GetInt("user_id")
diff --git a/internal/base/base.go b/internal/base/base.go
index f6956da..12f01df 100644
--- a/internal/base/base.go
+++ b/internal/base/base.go
@@ -6,6 +6,7 @@
 package base
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"sync"
@@ -30,6 +31,7 @@ const (
 	RetryQueue      = "asynq:retry"                  // ZSET
 	DeadQueue       = "asynq:dead"                   // ZSET
 	InProgressQueue = "asynq:in_progress"            // LIST
+	CancelChannel   = "asynq:cancel"                 // PubSub channel
 )
 
 // QueueKey returns a redis key string for the given queue name.
@@ -77,13 +79,19 @@ type TaskMessage struct {
 
 	// ErrorMsg holds the error message from the last failure.
 	ErrorMsg string
+
+	// Timeout specifies how long a task may run.
+	// The string value should be compatible with time.Duration.ParseDuration.
+	//
+	// Zero means no limit.
+	Timeout string
 }
 
 // ProcessInfo holds information about running background worker process.
 type ProcessInfo struct {
 	mu                sync.Mutex
 	Concurrency       int
-	Queues            map[string]uint
+	Queues            map[string]int
 	StrictPriority    bool
 	PID               int
 	Host              string
@@ -93,7 +101,7 @@ type ProcessInfo struct {
 }
 
 // NewProcessInfo returns a new instance of ProcessInfo.
-func NewProcessInfo(host string, pid, concurrency int, queues map[string]uint, strict bool) *ProcessInfo {
+func NewProcessInfo(host string, pid, concurrency int, queues map[string]int, strict bool) *ProcessInfo {
 	return &ProcessInfo{
 		Host:           host,
 		PID:            pid,
@@ -123,3 +131,50 @@ func (p *ProcessInfo) IncrActiveWorkerCount(delta int) {
 	defer p.mu.Unlock()
 	p.ActiveWorkerCount += delta
 }
+
+// Cancelations is a collection that holds cancel functions for all in-progress tasks.
+//
+// Its methods are safe to be used in multiple goroutines.
+type Cancelations struct {
+	mu          sync.Mutex
+	cancelFuncs map[string]context.CancelFunc
+}
+
+// NewCancelations returns a Cancelations instance.
+func NewCancelations() *Cancelations {
+	return &Cancelations{
+		cancelFuncs: make(map[string]context.CancelFunc),
+	}
+}
+
+// Add adds a new cancel func to the collection.
+func (c *Cancelations) Add(id string, fn context.CancelFunc) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.cancelFuncs[id] = fn
+}
+
+// Delete deletes a cancel func from the collection given an id.
+func (c *Cancelations) Delete(id string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.cancelFuncs, id)
+}
+
+// Get returns a cancel func given an id.
+func (c *Cancelations) Get(id string) context.CancelFunc {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.cancelFuncs[id]
+}
+
+// GetAll returns all cancel funcs.
+func (c *Cancelations) GetAll() []context.CancelFunc {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	var res []context.CancelFunc
+	for _, fn := range c.cancelFuncs {
+		res = append(res, fn)
+	}
+	return res
+}
diff --git a/internal/rdb/inspect.go b/internal/rdb/inspect.go
index 578b636..d0f4be7 100644
--- a/internal/rdb/inspect.go
+++ b/internal/rdb/inspect.go
@@ -245,10 +245,10 @@ func reverse(x []string) {
 // for the list operation.
 type Pagination struct {
 	// Number of items in the page.
-	Size uint
+	Size int
 
 	// Page number starting from zero.
-	Page uint
+	Page int
 }
 
 func (p Pagination) start() int64 {
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index b8e1259..3c166da 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -410,3 +410,19 @@ func (r *RDB) ClearProcessInfo(ps *base.ProcessInfo) error {
 	key := base.ProcessInfoKey(ps.Host, ps.PID)
 	return clearProcessInfoCmd.Run(r.client, []string{base.AllProcesses, key}).Err()
 }
+
+// CancelationPubSub returns a pubsub for cancelation messages.
+func (r *RDB) CancelationPubSub() (*redis.PubSub, error) {
+	pubsub := r.client.Subscribe(base.CancelChannel)
+	_, err := pubsub.Receive()
+	if err != nil {
+		return nil, err
+	}
+	return pubsub, nil
+}
+
+// PublishCancelation publish cancelation message to all subscribers.
+// The message is the ID for the task to be canceled.
+func (r *RDB) PublishCancelation(id string) error {
+	return r.client.Publish(base.CancelChannel, id).Err()
+}
diff --git a/processor.go b/processor.go
index 738128c..6bc4b1c 100644
--- a/processor.go
+++ b/processor.go
@@ -5,6 +5,7 @@
 package asynq
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"sort"
@@ -23,7 +24,7 @@ type processor struct {
 
 	handler Handler
 
-	queueConfig map[string]uint
+	queueConfig map[string]int
 
 	// orderedQueues is set only in strict-priority mode.
 	orderedQueues []string
@@ -50,12 +51,15 @@ type processor struct {
 
 	// quit channel communicates to the in-flight worker goroutines to stop.
 	quit chan struct{}
+
+	// cancelations is a set of cancel functions for all in-progress tasks.
+	cancelations *base.Cancelations
 }
 
 type retryDelayFunc func(n int, err error, task *Task) time.Duration
 
 // newProcessor constructs a new processor.
-func newProcessor(r *rdb.RDB, pinfo *base.ProcessInfo, fn retryDelayFunc, syncRequestCh chan<- *syncRequest) *processor {
+func newProcessor(r *rdb.RDB, pinfo *base.ProcessInfo, fn retryDelayFunc, syncRequestCh chan<- *syncRequest, cancelations *base.Cancelations) *processor {
 	qcfg := normalizeQueueCfg(pinfo.Queues)
 	orderedQueues := []string(nil)
 	if pinfo.StrictPriority {
@@ -68,12 +72,13 @@ func newProcessor(r *rdb.RDB, pinfo *base.ProcessInfo, fn retryDelayFunc, syncRe
 		orderedQueues:  orderedQueues,
 		retryDelayFunc: fn,
 		syncRequestCh:  syncRequestCh,
+		cancelations:   cancelations,
 		errLogLimiter:  rate.NewLimiter(rate.Every(3*time.Second), 1),
 		sema:           make(chan struct{}, pinfo.Concurrency),
 		done:           make(chan struct{}),
 		abort:          make(chan struct{}),
 		quit:           make(chan struct{}),
-		handler:        HandlerFunc(func(t *Task) error { return fmt.Errorf("handler not set") }),
+		handler:        HandlerFunc(func(ctx context.Context, t *Task) error { return fmt.Errorf("handler not set") }),
 	}
 }
 
@@ -98,6 +103,12 @@ func (p *processor) terminate() {
 	const timeout = 8 * time.Second
 	time.AfterFunc(timeout, func() { close(p.quit) })
 	logger.info("Waiting for all workers to finish...")
+
+	// send cancellation signal to all in-progress task handlers
+	for _, cancel := range p.cancelations.GetAll() {
+		cancel()
+	}
+
 	// block until all workers have released the token
 	for i := 0; i < cap(p.sema); i++ {
 		p.sema <- struct{}{}
@@ -160,8 +171,11 @@ func (p *processor) exec() {
 
 			resCh := make(chan error, 1)
 			task := NewTask(msg.Type, msg.Payload)
+			ctx, cancel := createContext(msg)
+			p.cancelations.Add(msg.ID.String(), cancel)
 			go func() {
-				resCh <- perform(p.handler, task)
+				resCh <- perform(ctx, task, p.handler)
+				p.cancelations.Delete(msg.ID.String())
 			}()
 
 			select {
@@ -282,13 +296,13 @@ func (p *processor) queues() []string {
 // perform calls the handler with the given task.
 // If the call returns without panic, it simply returns the value,
 // otherwise, it recovers from panic and returns an error.
-func perform(h Handler, task *Task) (err error) {
+func perform(ctx context.Context, task *Task, h Handler) (err error) {
 	defer func() {
 		if x := recover(); x != nil {
 			err = fmt.Errorf("panic: %v", x)
 		}
 	}()
-	return h.ProcessTask(task)
+	return h.ProcessTask(ctx, task)
 }
 
 // uniq dedupes elements and returns a slice of unique names of length l.
@@ -310,7 +324,7 @@ func uniq(names []string, l int) []string {
 
 // sortByPriority returns a list of queue names sorted by
 // their priority level in descending order.
-func sortByPriority(qcfg map[string]uint) []string {
+func sortByPriority(qcfg map[string]int) []string {
 	var queues []*queue
 	for qname, n := range qcfg {
 		queues = append(queues, &queue{qname, n})
@@ -325,7 +339,7 @@ func sortByPriority(qcfg map[string]uint) []string {
 
 type queue struct {
 	name     string
-	priority uint
+	priority int
 }
 
 type byPriority []*queue
@@ -336,21 +350,21 @@ func (x byPriority) Swap(i, j int)      { x[i], x[j] = x[j], x[i] }
 
 // normalizeQueueCfg divides priority numbers by their
 // greatest common divisor.
-func normalizeQueueCfg(queueCfg map[string]uint) map[string]uint {
-	var xs []uint
+func normalizeQueueCfg(queueCfg map[string]int) map[string]int {
+	var xs []int
 	for _, x := range queueCfg {
 		xs = append(xs, x)
 	}
 	d := gcd(xs...)
-	res := make(map[string]uint)
+	res := make(map[string]int)
 	for q, x := range queueCfg {
 		res[q] = x / d
 	}
 	return res
 }
 
-func gcd(xs ...uint) uint {
-	fn := func(x, y uint) uint {
+func gcd(xs ...int) int {
+	fn := func(x, y int) int {
 		for y > 0 {
 			x, y = y, x%y
 		}
@@ -365,3 +379,16 @@ func gcd(xs ...uint) uint {
 	}
 	return res
 }
+
+// createContext returns a context and cancel function for a given task message.
+func createContext(msg *base.TaskMessage) (context.Context, context.CancelFunc) {
+	timeout, err := time.ParseDuration(msg.Timeout)
+	if err != nil {
+		logger.error("cannot parse timeout duration for %+v", msg)
+		return context.WithCancel(context.Background())
+	}
+	if timeout == 0 {
+		return context.WithCancel(context.Background())
+	}
+	return context.WithTimeout(context.Background(), timeout)
+}
diff --git a/scheduler.go b/scheduler.go
index 8c59ebe..7276f4c 100644
--- a/scheduler.go
+++ b/scheduler.go
@@ -23,7 +23,7 @@ type scheduler struct {
 	qnames []string
 }
 
-func newScheduler(r *rdb.RDB, avgInterval time.Duration, qcfg map[string]uint) *scheduler {
+func newScheduler(r *rdb.RDB, avgInterval time.Duration, qcfg map[string]int) *scheduler {
 	var qnames []string
 	for q := range qcfg {
 		qnames = append(qnames, q)
diff --git a/subscriber.go b/subscriber.go
new file mode 100644
index 0000000..220116c
--- /dev/null
+++ b/subscriber.go
@@ -0,0 +1,58 @@
+// 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 asynq
+
+import (
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/rdb"
+)
+
+type subscriber struct {
+	rdb *rdb.RDB
+
+	// channel to communicate back to the long running "subscriber" goroutine.
+	done chan struct{}
+
+	// cancelations hold cancel functions for all in-progress tasks.
+	cancelations *base.Cancelations
+}
+
+func newSubscriber(rdb *rdb.RDB, cancelations *base.Cancelations) *subscriber {
+	return &subscriber{
+		rdb:          rdb,
+		done:         make(chan struct{}),
+		cancelations: cancelations,
+	}
+}
+
+func (s *subscriber) terminate() {
+	logger.info("Subscriber shutting down...")
+	// Signal the subscriber goroutine to stop.
+	s.done <- struct{}{}
+}
+
+func (s *subscriber) start() {
+	pubsub, err := s.rdb.CancelationPubSub()
+	cancelCh := pubsub.Channel()
+	if err != nil {
+		logger.error("cannot subscribe to cancelation channel: %v", err)
+		return
+	}
+	go func() {
+		for {
+			select {
+			case <-s.done:
+				pubsub.Close()
+				logger.info("Subscriber done")
+				return
+			case msg := <-cancelCh:
+				cancel := s.cancelations.Get(msg.Payload)
+				if cancel != nil {
+					cancel()
+				}
+			}
+		}
+	}()
+}
diff --git a/tools/asynqmon/README.md b/tools/asynqmon/README.md
index 56dfcc4..49f1b2c 100644
--- a/tools/asynqmon/README.md
+++ b/tools/asynqmon/README.md
@@ -13,6 +13,7 @@ Asynqmon is a command line tool to monitor the tasks managed by `asynq` package.
   - [Enqueue](#enqueue)
   - [Delete](#delete)
   - [Kill](#kill)
+  - [Cancel](#cancel)
 - [Config File](#config-file)
 
 ## Installation
@@ -133,6 +134,18 @@ Example:
 
 Running the above command will move all **Retry** tasks to **Dead** state.
 
+### Cancel
+
+Command `cancel` takes a task ID and sends a cancelation signal to the goroutine processing the specified task.  
+You can obtain the task ID by running `ls` command.
+
+The task should be in "in-progress" state.
+Handler implementation needs to be context aware in order to actually stop processing.
+
+Example:
+
+    asynqmon cancel bnogo8gt6toe23vhef0g
+
 ## Config File
 
 You can use a config file to set default values for the flags.
diff --git a/tools/asynqmon/cmd/cancel.go b/tools/asynqmon/cmd/cancel.go
new file mode 100644
index 0000000..772e8d1
--- /dev/null
+++ b/tools/asynqmon/cmd/cancel.go
@@ -0,0 +1,53 @@
+// 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"
+)
+
+// cancelCmd represents the cancel command
+var cancelCmd = &cobra.Command{
+	Use:   "cancel [task id]",
+	Short: "Sends a cancelation signal to the goroutine processing the specified task",
+	Long: `Cancel (asynqmon cancel) will send a cancelation signal to the goroutine processing 
+the specified task. 
+
+The command takes one argument which specifies the task to cancel.
+The task should be in in-progress state.
+Identifier for a task should be obtained by running "asynqmon ls" command.
+
+Handler implementation needs to be context aware for cancelation signal to
+actually cancel the processing.
+
+Example: asynqmon cancel bnogo8gt6toe23vhef0g`,
+	Args: cobra.ExactArgs(1),
+	Run:  cancel,
+}
+
+func init() {
+	rootCmd.AddCommand(cancelCmd)
+}
+
+func cancel(cmd *cobra.Command, args []string) {
+	r := rdb.NewRDB(redis.NewClient(&redis.Options{
+		Addr:     viper.GetString("uri"),
+		DB:       viper.GetInt("db"),
+		Password: viper.GetString("password"),
+	}))
+
+	err := r.PublishCancelation(args[0])
+	if err != nil {
+		fmt.Printf("could not send cancelation signal: %v\n", err)
+		os.Exit(1)
+	}
+	fmt.Printf("Successfully sent cancelation siganl for task %s\n", args[0])
+}
diff --git a/tools/asynqmon/cmd/ls.go b/tools/asynqmon/cmd/ls.go
index a4b85ee..3bec832 100644
--- a/tools/asynqmon/cmd/ls.go
+++ b/tools/asynqmon/cmd/ls.go
@@ -44,16 +44,24 @@ asynqmon ls enqueued:critical -> List tasks from critical queue
 }
 
 // Flags
-var pageSize uint
-var pageNum uint
+var pageSize int
+var pageNum int
 
 func init() {
 	rootCmd.AddCommand(lsCmd)
-	lsCmd.Flags().UintVar(&pageSize, "size", 30, "page size")
-	lsCmd.Flags().UintVar(&pageNum, "page", 0, "page number - zero indexed (default 0)")
+	lsCmd.Flags().IntVar(&pageSize, "size", 30, "page size")
+	lsCmd.Flags().IntVar(&pageNum, "page", 0, "page number - zero indexed (default 0)")
 }
 
 func ls(cmd *cobra.Command, args []string) {
+	if pageSize < 0 {
+		fmt.Println("page size cannot be negative.")
+		os.Exit(1)
+	}
+	if pageNum < 0 {
+		fmt.Println("page number cannot be negative.")
+		os.Exit(1)
+	}
 	c := redis.NewClient(&redis.Options{
 		Addr:     viper.GetString("uri"),
 		DB:       viper.GetInt("db"),
diff --git a/tools/asynqmon/cmd/ps.go b/tools/asynqmon/cmd/ps.go
index ef25876..7761886 100644
--- a/tools/asynqmon/cmd/ps.go
+++ b/tools/asynqmon/cmd/ps.go
@@ -87,11 +87,11 @@ func timeAgo(since time.Time) string {
 	return fmt.Sprintf("%v ago", d)
 }
 
-func formatQueues(qmap map[string]uint) string {
+func formatQueues(qmap map[string]int) string {
 	// sort queues by priority and name
 	type queue struct {
 		name     string
-		priority uint
+		priority int
 	}
 	var queues []*queue
 	for qname, p := range qmap {


Test patch:
diff --git a/background_test.go b/background_test.go
index 2acebbc..32aaeb2 100644
--- a/background_test.go
+++ b/background_test.go
@@ -5,6 +5,7 @@
 package asynq
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -27,7 +28,7 @@ func TestBackground(t *testing.T) {
 	})
 
 	// no-op handler
-	h := func(task *Task) error {
+	h := func(ctx context.Context, task *Task) error {
 		return nil
 	}
 
@@ -42,16 +43,16 @@ func TestBackground(t *testing.T) {
 
 func TestGCD(t *testing.T) {
 	tests := []struct {
-		input []uint
-		want  uint
+		input []int
+		want  int
 	}{
-		{[]uint{6, 2, 12}, 2},
-		{[]uint{3, 3, 3}, 3},
-		{[]uint{6, 3, 1}, 1},
-		{[]uint{1}, 1},
-		{[]uint{1, 0, 2}, 1},
-		{[]uint{8, 0, 4}, 4},
-		{[]uint{9, 12, 18, 30}, 3},
+		{[]int{6, 2, 12}, 2},
+		{[]int{3, 3, 3}, 3},
+		{[]int{6, 3, 1}, 1},
+		{[]int{1}, 1},
+		{[]int{1, 0, 2}, 1},
+		{[]int{8, 0, 4}, 4},
+		{[]int{9, 12, 18, 30}, 3},
 	}
 
 	for _, tc := range tests {
@@ -64,46 +65,46 @@ func TestGCD(t *testing.T) {
 
 func TestNormalizeQueueCfg(t *testing.T) {
 	tests := []struct {
-		input map[string]uint
-		want  map[string]uint
+		input map[string]int
+		want  map[string]int
 	}{
 		{
-			input: map[string]uint{
+			input: map[string]int{
 				"high":    100,
 				"default": 20,
 				"low":     5,
 			},
-			want: map[string]uint{
+			want: map[string]int{
 				"high":    20,
 				"default": 4,
 				"low":     1,
 			},
 		},
 		{
-			input: map[string]uint{
+			input: map[string]int{
 				"default": 10,
 			},
-			want: map[string]uint{
+			want: map[string]int{
 				"default": 1,
 			},
 		},
 		{
-			input: map[string]uint{
+			input: map[string]int{
 				"critical": 5,
 				"default":  1,
 			},
-			want: map[string]uint{
+			want: map[string]int{
 				"critical": 5,
 				"default":  1,
 			},
 		},
 		{
-			input: map[string]uint{
+			input: map[string]int{
 				"critical": 6,
 				"default":  3,
 				"low":      0,
 			},
-			want: map[string]uint{
+			want: map[string]int{
 				"critical": 2,
 				"default":  1,
 				"low":      0,
diff --git a/benchmark_test.go b/benchmark_test.go
index 580bab7..5988adb 100644
--- a/benchmark_test.go
+++ b/benchmark_test.go
@@ -5,6 +5,7 @@
 package asynq
 
 import (
+	"context"
 	"fmt"
 	"math/rand"
 	"sync"
@@ -37,7 +38,7 @@ func BenchmarkEndToEndSimple(b *testing.B) {
 
 		var wg sync.WaitGroup
 		wg.Add(count)
-		handler := func(t *Task) error {
+		handler := func(ctx context.Context, t *Task) error {
 			wg.Done()
 			return nil
 		}
@@ -82,7 +83,7 @@ func BenchmarkEndToEnd(b *testing.B) {
 
 		var wg sync.WaitGroup
 		wg.Add(count * 2)
-		handler := func(t *Task) error {
+		handler := func(ctx context.Context, t *Task) error {
 			// randomly fail 1% of tasks
 			if rand.Intn(100) == 1 {
 				return fmt.Errorf(":(")
@@ -119,7 +120,7 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 		client := NewClient(redis)
 		bg := NewBackground(redis, &Config{
 			Concurrency: 10,
-			Queues: map[string]uint{
+			Queues: map[string]int{
 				"high":    6,
 				"default": 3,
 				"low":     1,
@@ -141,7 +142,7 @@ func BenchmarkEndToEndMultipleQueues(b *testing.B) {
 
 		var wg sync.WaitGroup
 		wg.Add(highCount + defaultCount + lowCount)
-		handler := func(t *Task) error {
+		handler := func(ctx context.Context, t *Task) error {
 			wg.Done()
 			return nil
 		}
diff --git a/client_test.go b/client_test.go
index 7024f5d..1bd19ef 100644
--- a/client_test.go
+++ b/client_test.go
@@ -42,6 +42,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   defaultMaxRetry,
 						Queue:   "default",
+						Timeout: time.Duration(0).String(),
 					},
 				},
 			},
@@ -60,6 +61,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   defaultMaxRetry,
 						Queue:   "default",
+						Timeout: time.Duration(0).String(),
 					},
 					Score: float64(time.Now().Add(2 * time.Hour).Unix()),
 				},
@@ -79,6 +81,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   3,
 						Queue:   "default",
+						Timeout: time.Duration(0).String(),
 					},
 				},
 			},
@@ -98,6 +101,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   0, // Retry count should be set to zero
 						Queue:   "default",
+						Timeout: time.Duration(0).String(),
 					},
 				},
 			},
@@ -118,6 +122,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   10, // Last option takes precedence
 						Queue:   "default",
+						Timeout: time.Duration(0).String(),
 					},
 				},
 			},
@@ -137,6 +142,7 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   defaultMaxRetry,
 						Queue:   "custom",
+						Timeout: time.Duration(0).String(),
 					},
 				},
 			},
@@ -156,6 +162,27 @@ func TestClient(t *testing.T) {
 						Payload: task.Payload.data,
 						Retry:   defaultMaxRetry,
 						Queue:   "high",
+						Timeout: time.Duration(0).String(),
+					},
+				},
+			},
+			wantScheduled: nil, // db is flushed in setup so zset does not exist hence nil
+		},
+		{
+			desc:      "Timeout option sets the timeout duration",
+			task:      task,
+			processAt: time.Now(),
+			opts: []Option{
+				Timeout(20 * time.Second),
+			},
+			wantEnqueued: map[string][]*base.TaskMessage{
+				"default": []*base.TaskMessage{
+					&base.TaskMessage{
+						Type:    task.Type,
+						Payload: task.Payload.data,
+						Retry:   defaultMaxRetry,
+						Queue:   "default",
+						Timeout: (20 * time.Second).String(),
 					},
 				},
 			},
diff --git a/heartbeat_test.go b/heartbeat_test.go
index d4fffb2..268ea76 100644
--- a/heartbeat_test.go
+++ b/heartbeat_test.go
@@ -23,10 +23,10 @@ func TestHeartbeater(t *testing.T) {
 		interval    time.Duration
 		host        string
 		pid         int
-		queues      map[string]uint
+		queues      map[string]int
 		concurrency int
 	}{
-		{time.Second, "some.address.ec2.aws.com", 45678, map[string]uint{"default": 1}, 10},
+		{time.Second, "some.address.ec2.aws.com", 45678, map[string]int{"default": 1}, 10},
 	}
 
 	timeCmpOpt := cmpopts.EquateApproxTime(10 * time.Millisecond)
diff --git a/internal/base/base_test.go b/internal/base/base_test.go
index 00be893..46050c0 100644
--- a/internal/base/base_test.go
+++ b/internal/base/base_test.go
@@ -82,7 +82,7 @@ func TestProcessInfoKey(t *testing.T) {
 
 // Note: Run this test with -race flag to check for data race.
 func TestProcessInfoSetter(t *testing.T) {
-	pi := NewProcessInfo("localhost", 1234, 8, map[string]uint{"default": 1}, false)
+	pi := NewProcessInfo("localhost", 1234, 8, map[string]int{"default": 1}, false)
 
 	var wg sync.WaitGroup
 
diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go
index 91422c0..cd671c2 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -317,8 +317,8 @@ func TestListEnqueuedPagination(t *testing.T) {
 	tests := []struct {
 		desc      string
 		qname     string
-		page      uint
-		size      uint
+		page      int
+		size      int
 		wantSize  int
 		wantFirst string
 		wantLast  string
@@ -418,8 +418,8 @@ func TestListInProgressPagination(t *testing.T) {
 
 	tests := []struct {
 		desc      string
-		page      uint
-		size      uint
+		page      int
+		size      int
 		wantSize  int
 		wantFirst string
 		wantLast  string
@@ -524,8 +524,8 @@ func TestListScheduledPagination(t *testing.T) {
 
 	tests := []struct {
 		desc      string
-		page      uint
-		size      uint
+		page      int
+		size      int
 		wantSize  int
 		wantFirst string
 		wantLast  string
@@ -667,8 +667,8 @@ func TestListRetryPagination(t *testing.T) {
 
 	tests := []struct {
 		desc      string
-		page      uint
-		size      uint
+		page      int
+		size      int
 		wantSize  int
 		wantFirst string
 		wantLast  string
@@ -800,8 +800,8 @@ func TestListDeadPagination(t *testing.T) {
 
 	tests := []struct {
 		desc      string
-		page      uint
-		size      uint
+		page      int
+		size      int
 		wantSize  int
 		wantFirst string
 		wantLast  string
@@ -2056,7 +2056,7 @@ func TestListProcesses(t *testing.T) {
 
 	ps1 := &base.ProcessInfo{
 		Concurrency:       10,
-		Queues:            map[string]uint{"default": 1},
+		Queues:            map[string]int{"default": 1},
 		Host:              "do.droplet1",
 		PID:               1234,
 		State:             "running",
@@ -2066,7 +2066,7 @@ func TestListProcesses(t *testing.T) {
 
 	ps2 := &base.ProcessInfo{
 		Concurrency:       20,
-		Queues:            map[string]uint{"email": 1},
+		Queues:            map[string]int{"email": 1},
 		Host:              "do.droplet2",
 		PID:               9876,
 		State:             "stopped",
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index 8b8d52e..accc52d 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -745,7 +745,7 @@ func TestReadWriteClearProcessInfo(t *testing.T) {
 	r := setup(t)
 	pinfo := &base.ProcessInfo{
 		Concurrency:       10,
-		Queues:            map[string]uint{"default": 2, "email": 5, "low": 1},
+		Queues:            map[string]int{"default": 2, "email": 5, "low": 1},
 		PID:               98765,
 		Host:              "localhost",
 		State:             "running",
diff --git a/processor_test.go b/processor_test.go
index 893a9f3..079bf6e 100644
--- a/processor_test.go
+++ b/processor_test.go
@@ -5,6 +5,7 @@
 package asynq
 
 import (
+	"context"
 	"fmt"
 	"sort"
 	"sync"
@@ -59,14 +60,15 @@ func TestProcessorSuccess(t *testing.T) {
 		// instantiate a new processor
 		var mu sync.Mutex
 		var processed []*Task
-		handler := func(task *Task) error {
+		handler := func(ctx context.Context, task *Task) error {
 			mu.Lock()
 			defer mu.Unlock()
 			processed = append(processed, task)
 			return nil
 		}
 		pi := base.NewProcessInfo("localhost", 1234, 10, defaultQueueConfig, false)
-		p := newProcessor(rdbClient, pi, defaultDelayFunc, nil)
+		cancelations := base.NewCancelations()
+		p := newProcessor(rdbClient, pi, defaultDelayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		p.start()
@@ -146,11 +148,12 @@ func TestProcessorRetry(t *testing.T) {
 		delayFunc := func(n int, e error, t *Task) time.Duration {
 			return tc.delay
 		}
-		handler := func(task *Task) error {
+		handler := func(ctx context.Context, task *Task) error {
 			return fmt.Errorf(errMsg)
 		}
 		pi := base.NewProcessInfo("localhost", 1234, 10, defaultQueueConfig, false)
-		p := newProcessor(rdbClient, pi, delayFunc, nil)
+		cancelations := base.NewCancelations()
+		p := newProcessor(rdbClient, pi, delayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		p.start()
@@ -189,11 +192,11 @@ func TestProcessorQueues(t *testing.T) {
 	})
 
 	tests := []struct {
-		queueCfg map[string]uint
+		queueCfg map[string]int
 		want     []string
 	}{
 		{
-			queueCfg: map[string]uint{
+			queueCfg: map[string]int{
 				"high":    6,
 				"default": 3,
 				"low":     1,
@@ -201,7 +204,7 @@ func TestProcessorQueues(t *testing.T) {
 			want: []string{"high", "default", "low"},
 		},
 		{
-			queueCfg: map[string]uint{
+			queueCfg: map[string]int{
 				"default": 1,
 			},
 			want: []string{"default"},
@@ -210,7 +213,8 @@ func TestProcessorQueues(t *testing.T) {
 
 	for _, tc := range tests {
 		pi := base.NewProcessInfo("localhost", 1234, 10, tc.queueCfg, false)
-		p := newProcessor(nil, pi, defaultDelayFunc, nil)
+		cancelations := base.NewCancelations()
+		p := newProcessor(nil, pi, defaultDelayFunc, nil, cancelations)
 		got := p.queues()
 		if diff := cmp.Diff(tc.want, got, sortOpt); diff != "" {
 			t.Errorf("with queue config: %v\n(*processor).queues() = %v, want %v\n(-want,+got):\n%s",
@@ -264,20 +268,21 @@ func TestProcessorWithStrictPriority(t *testing.T) {
 		// instantiate a new processor
 		var mu sync.Mutex
 		var processed []*Task
-		handler := func(task *Task) error {
+		handler := func(ctx context.Context, task *Task) error {
 			mu.Lock()
 			defer mu.Unlock()
 			processed = append(processed, task)
 			return nil
 		}
-		queueCfg := map[string]uint{
+		queueCfg := map[string]int{
 			"critical":            3,
 			base.DefaultQueueName: 2,
 			"low":                 1,
 		}
 		// Note: Set concurrency to 1 to make sure tasks are processed one at a time.
 		pi := base.NewProcessInfo("localhost", 1234, 1 /*concurrency */, queueCfg, true /* strict */)
-		p := newProcessor(rdbClient, pi, defaultDelayFunc, nil)
+		cancelations := base.NewCancelations()
+		p := newProcessor(rdbClient, pi, defaultDelayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		p.start()
@@ -303,7 +308,7 @@ func TestPerform(t *testing.T) {
 	}{
 		{
 			desc: "handler returns nil",
-			handler: func(t *Task) error {
+			handler: func(ctx context.Context, t *Task) error {
 				return nil
 			},
 			task:    NewTask("gen_thumbnail", map[string]interface{}{"src": "some/img/path"}),
@@ -311,7 +316,7 @@ func TestPerform(t *testing.T) {
 		},
 		{
 			desc: "handler returns error",
-			handler: func(t *Task) error {
+			handler: func(ctx context.Context, t *Task) error {
 				return fmt.Errorf("something went wrong")
 			},
 			task:    NewTask("gen_thumbnail", map[string]interface{}{"src": "some/img/path"}),
@@ -319,7 +324,7 @@ func TestPerform(t *testing.T) {
 		},
 		{
 			desc: "handler panics",
-			handler: func(t *Task) error {
+			handler: func(ctx context.Context, t *Task) error {
 				panic("something went terribly wrong")
 			},
 			task:    NewTask("gen_thumbnail", map[string]interface{}{"src": "some/img/path"}),
@@ -328,7 +333,7 @@ func TestPerform(t *testing.T) {
 	}
 
 	for _, tc := range tests {
-		got := perform(tc.handler, tc.task)
+		got := perform(context.Background(), tc.task, tc.handler)
 		if !tc.wantErr && got != nil {
 			t.Errorf("%s: perform() = %v, want nil", tc.desc, got)
 			continue
diff --git a/subscriber_test.go b/subscriber_test.go
new file mode 100644
index 0000000..2b7f5a0
--- /dev/null
+++ b/subscriber_test.go
@@ -0,0 +1,57 @@
+// 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 asynq
+
+import (
+	"testing"
+	"time"
+
+	"github.com/hibiken/asynq/internal/base"
+	"github.com/hibiken/asynq/internal/rdb"
+)
+
+func TestSubscriber(t *testing.T) {
+	r := setup(t)
+	rdbClient := rdb.NewRDB(r)
+
+	tests := []struct {
+		registeredID string // ID for which cancel func is registered
+		publishID    string // ID to be published
+		wantCalled   bool   // whether cancel func should be called
+	}{
+		{"abc123", "abc123", true},
+		{"abc456", "abc123", false},
+	}
+
+	for _, tc := range tests {
+		called := false
+		fakeCancelFunc := func() {
+			called = true
+		}
+		cancelations := base.NewCancelations()
+		cancelations.Add(tc.registeredID, fakeCancelFunc)
+
+		subscriber := newSubscriber(rdbClient, cancelations)
+		subscriber.start()
+
+		if err := rdbClient.PublishCancelation(tc.publishID); err != nil {
+			subscriber.terminate()
+			t.Fatalf("could not publish cancelation message: %v", err)
+		}
+
+		// allow for redis to publish message
+		time.Sleep(time.Second)
+
+		if called != tc.wantCalled {
+			if tc.wantCalled {
+				t.Errorf("fakeCancelFunc was not called, want the function to be called")
+			} else {
+				t.Errorf("fakeCancelFunc was called, want the function to not be called")
+			}
+		}
+
+		subscriber.terminate()
+	}
+}