Repository: hibiken/asynq
Pull request #93 (2020-02-21 15:20:05)

Record worker started time
Prework for #39 

codecov[bot]: # [Codecov](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=h1) Report
> Merging [#93](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=desc) into [master](https://codecov.io/gh/hibiken/asynq/commit/8540172306a39f25dcfe5426b305537e9d4ff38f&el=desc) will **increase** coverage by `2.81%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/hibiken/asynq/pull/93/graphs/tree.svg?width=650&height=150&src=pr&token=BAKLSWL2G3)](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master      #93      +/-   ##
==========================================
+ Coverage   75.56%   78.38%   +2.81%     
==========================================
  Files          13       13              
  Lines        1101     1161      +60     
==========================================
+ Hits          832      910      +78     
+ Misses        200      185      -15     
+ Partials       69       66       -3     
```


| [Impacted Files](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [internal/rdb/inspect.go](https://codecov.io/gh/hibiken/asynq/pull/93/diff?src=pr&el=tree#diff-aW50ZXJuYWwvcmRiL2luc3BlY3QuZ28=) | `73.76% <0.00%> (-2.92%)` | :arrow_down: |
| [processor.go](https://codecov.io/gh/hibiken/asynq/pull/93/diff?src=pr&el=tree#diff-cHJvY2Vzc29yLmdv) | `78.39% <0.00%> (-0.91%)` | :arrow_down: |
| [internal/base/base.go](https://codecov.io/gh/hibiken/asynq/pull/93/diff?src=pr&el=tree#diff-aW50ZXJuYWwvYmFzZS9iYXNlLmdv) | `98.98% <0.00%> (+77.36%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=footer). Last update [f9a6c61...f8a2861](https://codecov.io/gh/hibiken/asynq/pull/93?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/background.go b/background.go
index e2e27eb..1a39598 100644
--- a/background.go
+++ b/background.go
@@ -34,8 +34,7 @@ type Background struct {
 	mu      sync.Mutex
 	running bool
 
-	// channel to send state updates.
-	stateCh chan<- string
+	ps *base.ProcessState
 
 	// wait group to wait for all goroutines to finish.
 	wg sync.WaitGroup
@@ -131,18 +130,17 @@ func NewBackground(r RedisConnOpt, cfg *Config) *Background {
 	pid := os.Getpid()
 
 	rdb := rdb.NewRDB(createRedisClient(r))
-	syncRequestCh := make(chan *syncRequest)
-	stateCh := make(chan string)
-	workerCh := make(chan int)
-	cancelations := base.NewCancelations()
-	syncer := newSyncer(syncRequestCh, 5*time.Second)
-	heartbeater := newHeartbeater(rdb, host, pid, n, queues, cfg.StrictPriority, 5*time.Second, stateCh, workerCh)
+	ps := base.NewProcessState(host, pid, n, queues, cfg.StrictPriority)
+	syncCh := make(chan *syncRequest)
+	cancels := base.NewCancelations()
+	syncer := newSyncer(syncCh, 5*time.Second)
+	heartbeater := newHeartbeater(rdb, ps, 5*time.Second)
 	scheduler := newScheduler(rdb, 5*time.Second, queues)
-	processor := newProcessor(rdb, queues, cfg.StrictPriority, n, delayFunc, syncRequestCh, workerCh, cancelations)
-	subscriber := newSubscriber(rdb, cancelations)
+	processor := newProcessor(rdb, ps, delayFunc, syncCh, cancels)
+	subscriber := newSubscriber(rdb, cancels)
 	return &Background{
-		stateCh:     stateCh,
 		rdb:         rdb,
+		ps:          ps,
 		scheduler:   scheduler,
 		processor:   processor,
 		syncer:      syncer,
@@ -194,7 +192,7 @@ func (bg *Background) Run(handler Handler) {
 		sig := <-sigs
 		if sig == syscall.SIGTSTP {
 			bg.processor.stop()
-			bg.stateCh <- "stopped"
+			bg.ps.SetStatus(base.StatusStopped)
 			continue
 		}
 		break
@@ -232,8 +230,7 @@ func (bg *Background) stop() {
 	// Note: The order of termination is important.
 	// Sender goroutines should be terminated before the receiver goroutines.
 	//
-	// processor -> syncer      (via syncRequestCh)
-	// processor -> heartbeater (via workerCh)
+	// processor -> syncer (via syncCh)
 	bg.scheduler.terminate()
 	bg.processor.terminate()
 	bg.syncer.terminate()
diff --git a/heartbeat.go b/heartbeat.go
index 9c78b74..556a219 100644
--- a/heartbeat.go
+++ b/heartbeat.go
@@ -17,29 +17,20 @@ import (
 type heartbeater struct {
 	rdb *rdb.RDB
 
-	pinfo *base.ProcessInfo
+	ps *base.ProcessState
 
 	// channel to communicate back to the long running "heartbeater" goroutine.
 	done chan struct{}
 
-	// channel to receive updates on process state.
-	stateCh <-chan string
-
-	// channel to recieve updates on workers count.
-	workerCh <-chan int
-
 	// interval between heartbeats.
 	interval time.Duration
 }
 
-func newHeartbeater(rdb *rdb.RDB, host string, pid, concurrency int, queues map[string]int, strict bool,
-	interval time.Duration, stateCh <-chan string, workerCh <-chan int) *heartbeater {
+func newHeartbeater(rdb *rdb.RDB, ps *base.ProcessState, interval time.Duration) *heartbeater {
 	return &heartbeater{
 		rdb:      rdb,
-		pinfo:    base.NewProcessInfo(host, pid, concurrency, queues, strict),
+		ps:       ps,
 		done:     make(chan struct{}),
-		stateCh:  stateCh,
-		workerCh: workerCh,
 		interval: interval,
 	}
 }
@@ -51,26 +42,20 @@ func (h *heartbeater) terminate() {
 }
 
 func (h *heartbeater) start(wg *sync.WaitGroup) {
-	h.pinfo.Started = time.Now()
-	h.pinfo.State = "running"
+	h.ps.SetStarted(time.Now())
+	h.ps.SetStatus(base.StatusRunning)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		h.beat()
-		timer := time.NewTimer(h.interval)
 		for {
 			select {
 			case <-h.done:
-				h.rdb.ClearProcessInfo(h.pinfo)
+				h.rdb.ClearProcessState(h.ps)
 				logger.info("Heartbeater done")
 				return
-			case state := <-h.stateCh:
-				h.pinfo.State = state
-			case delta := <-h.workerCh:
-				h.pinfo.ActiveWorkerCount += delta
-			case <-timer.C:
+			case <-time.After(h.interval):
 				h.beat()
-				timer.Reset(h.interval)
 			}
 		}
 	}()
@@ -79,7 +64,7 @@ func (h *heartbeater) start(wg *sync.WaitGroup) {
 func (h *heartbeater) beat() {
 	// Note: Set TTL to be long enough so that it won't expire before we write again
 	// and short enough to expire quickly once the process is shut down or killed.
-	err := h.rdb.WriteProcessInfo(h.pinfo, h.interval*2)
+	err := h.rdb.WriteProcessState(h.ps, h.interval*2)
 	if err != nil {
 		logger.error("could not write heartbeat data: %v", err)
 	}
diff --git a/internal/base/base.go b/internal/base/base.go
index d616901..2f9ab34 100644
--- a/internal/base/base.go
+++ b/internal/base/base.go
@@ -20,8 +20,10 @@ const DefaultQueueName = "default"
 
 // Redis keys
 const (
-	psPrefix        = "asynq:ps:"                    // HASH
 	AllProcesses    = "asynq:ps"                     // ZSET
+	psPrefix        = "asynq:ps:"                    // STRING - asynq:ps:<host>:<pid>
+	AllWorkers      = "asynq:workers"                // ZSET
+	workersPrefix   = "asynq:workers:"               // HASH   - asynq:workers:<host:<pid>
 	processedPrefix = "asynq:processed:"             // STRING - asynq:processed:<yyyy-mm-dd>
 	failurePrefix   = "asynq:failure:"               // STRING - asynq:failure:<yyyy-mm-dd>
 	QueuePrefix     = "asynq:queues:"                // LIST   - asynq:queues:<qname>
@@ -34,28 +36,31 @@ const (
 	CancelChannel   = "asynq:cancel"                 // PubSub channel
 )
 
-// QueueKey returns a redis key string for the given queue name.
+// QueueKey returns a redis key for the given queue name.
 func QueueKey(qname string) string {
 	return QueuePrefix + strings.ToLower(qname)
 }
 
-// ProcessedKey returns a redis key string for processed count
-// for the given day.
+// ProcessedKey returns a redis key for processed count for the given day.
 func ProcessedKey(t time.Time) string {
 	return processedPrefix + t.UTC().Format("2006-01-02")
 }
 
-// FailureKey returns a redis key string for failure count
-// for the given day.
+// FailureKey returns a redis key for failure count for the given day.
 func FailureKey(t time.Time) string {
 	return failurePrefix + t.UTC().Format("2006-01-02")
 }
 
-// ProcessInfoKey returns a redis key string for process info.
+// ProcessInfoKey returns a redis key for process info.
 func ProcessInfoKey(hostname string, pid int) string {
 	return fmt.Sprintf("%s%s:%d", psPrefix, hostname, pid)
 }
 
+// WorkersKey returns a redis key for the workers given hostname and pid.
+func WorkersKey(hostname string, pid int) string {
+	return fmt.Sprintf("%s%s:%d", workersPrefix, hostname, pid)
+}
+
 // TaskMessage is the internal representation of a task with additional metadata fields.
 // Serialized data of this type gets written to redis.
 type TaskMessage struct {
@@ -87,32 +92,171 @@ type TaskMessage struct {
 	Timeout string
 }
 
-// ProcessInfo holds information about running background worker process.
+// ProcessState holds process level information.
+//
+// ProcessStates are safe for concurrent use by multiple goroutines.
+type ProcessState struct {
+	mu             sync.Mutex // guards all data fields
+	concurrency    int
+	queues         map[string]int
+	strictPriority bool
+	pid            int
+	host           string
+	status         PStatus
+	started        time.Time
+	workers        map[string]*workerStats
+}
+
+// PStatus represents status of a process.
+type PStatus int
+
+const (
+	// StatusIdle indicates process is in idle state.
+	StatusIdle PStatus = iota
+
+	// StatusRunning indicates process is up and processing tasks.
+	StatusRunning
+
+	// StatusStopped indicates process is up but not processing new tasks.
+	StatusStopped
+)
+
+var statuses = []string{
+	"idle",
+	"running",
+	"stopped",
+}
+
+func (s PStatus) String() string {
+	if StatusIdle <= s && s <= StatusStopped {
+		return statuses[s]
+	}
+	return "unknown status"
+}
+
+type workerStats struct {
+	msg     *TaskMessage
+	started time.Time
+}
+
+// NewProcessState returns a new instance of ProcessState.
+func NewProcessState(host string, pid, concurrency int, queues map[string]int, strict bool) *ProcessState {
+	return &ProcessState{
+		host:           host,
+		pid:            pid,
+		concurrency:    concurrency,
+		queues:         cloneQueueConfig(queues),
+		strictPriority: strict,
+		status:         StatusIdle,
+		workers:        make(map[string]*workerStats),
+	}
+}
+
+// SetStatus updates the state of process.
+func (ps *ProcessState) SetStatus(status PStatus) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	ps.status = status
+}
+
+// SetStarted records when the process started processing.
+func (ps *ProcessState) SetStarted(t time.Time) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	ps.started = t
+}
+
+// AddWorkerStats records when a worker started and which task it's processing.
+func (ps *ProcessState) AddWorkerStats(msg *TaskMessage, started time.Time) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	ps.workers[msg.ID.String()] = &workerStats{msg, started}
+}
+
+// DeleteWorkerStats removes a worker's entry from the process state.
+func (ps *ProcessState) DeleteWorkerStats(msg *TaskMessage) {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	delete(ps.workers, msg.ID.String())
+}
+
+// Get returns current state of process as a ProcessInfo.
+func (ps *ProcessState) Get() *ProcessInfo {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	return &ProcessInfo{
+		Host:              ps.host,
+		PID:               ps.pid,
+		Concurrency:       ps.concurrency,
+		Queues:            cloneQueueConfig(ps.queues),
+		StrictPriority:    ps.strictPriority,
+		Status:            ps.status.String(),
+		Started:           ps.started,
+		ActiveWorkerCount: len(ps.workers),
+	}
+}
+
+// GetWorkers returns a list of currently running workers' info.
+func (ps *ProcessState) GetWorkers() []*WorkerInfo {
+	ps.mu.Lock()
+	defer ps.mu.Unlock()
+	var res []*WorkerInfo
+	for _, w := range ps.workers {
+		res = append(res, &WorkerInfo{
+			Host:    ps.host,
+			PID:     ps.pid,
+			ID:      w.msg.ID,
+			Type:    w.msg.Type,
+			Queue:   w.msg.Queue,
+			Payload: clonePayload(w.msg.Payload),
+			Started: w.started,
+		})
+	}
+	return res
+}
+
+func cloneQueueConfig(qcfg map[string]int) map[string]int {
+	res := make(map[string]int)
+	for qname, n := range qcfg {
+		res[qname] = n
+	}
+	return res
+}
+
+func clonePayload(payload map[string]interface{}) map[string]interface{} {
+	res := make(map[string]interface{})
+	for k, v := range payload {
+		res[k] = v
+	}
+	return res
+}
+
+// ProcessInfo holds information about a running background worker process.
 type ProcessInfo struct {
+	Host              string
+	PID               int
 	Concurrency       int
 	Queues            map[string]int
 	StrictPriority    bool
-	PID               int
-	Host              string
-	State             string
+	Status            string
 	Started           time.Time
 	ActiveWorkerCount int
 }
 
-// NewProcessInfo returns a new instance of ProcessInfo.
-func NewProcessInfo(host string, pid, concurrency int, queues map[string]int, strict bool) *ProcessInfo {
-	return &ProcessInfo{
-		Host:           host,
-		PID:            pid,
-		Concurrency:    concurrency,
-		Queues:         queues,
-		StrictPriority: strict,
-	}
+// WorkerInfo holds information about a running worker.
+type WorkerInfo struct {
+	Host    string
+	PID     int
+	ID      xid.ID
+	Type    string
+	Queue   string
+	Payload map[string]interface{}
+	Started time.Time
 }
 
 // Cancelations is a collection that holds cancel functions for all in-progress tasks.
 //
-// Its methods are safe to be used in multiple goroutines.
+// Cancelations are safe for concurrent use by multipel goroutines.
 type Cancelations struct {
 	mu          sync.Mutex
 	cancelFuncs map[string]context.CancelFunc
@@ -140,10 +284,11 @@ func (c *Cancelations) Delete(id string) {
 }
 
 // Get returns a cancel func given an id.
-func (c *Cancelations) Get(id string) context.CancelFunc {
+func (c *Cancelations) Get(id string) (fn context.CancelFunc, ok bool) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	return c.cancelFuncs[id]
+	fn, ok = c.cancelFuncs[id]
+	return fn, ok
 }
 
 // GetAll returns all cancel funcs.
diff --git a/internal/rdb/rdb.go b/internal/rdb/rdb.go
index 3c166da..faf12ed 100644
--- a/internal/rdb/rdb.go
+++ b/internal/rdb/rdb.go
@@ -359,56 +359,70 @@ func (r *RDB) forwardSingle(src, dst string) error {
 		[]string{src, dst}, now).Err()
 }
 
-// KEYS[1] -> asynq:ps
-// KEYS[2] -> asynq:ps:<host:pid>
-// ARGV[1] -> expiration time
-// ARGV[2] -> TTL in seconds
-// ARGV[3] -> process info
+// KEYS[1]  -> asynq:ps:<host:pid>
+// KEYS[2]  -> asynq:ps
+// KEYS[3]  -> asynq:workers<host:pid>
+// keys[4]  -> asynq:workers
+// ARGV[1]  -> expiration time
+// ARGV[2]  -> TTL in seconds
+// ARGV[3]  -> process info
+// ARGV[4:] -> alternate key-value pair of (worker id, worker data)
+// Note: Add key to ZSET with expiration time as score.
+// ref: https://github.com/antirez/redis/issues/135#issuecomment-2361996
 var writeProcessInfoCmd = redis.NewScript(`
-redis.call("ZADD", KEYS[1], ARGV[1], KEYS[2])
-redis.call("SETEX", KEYS[2], ARGV[2], ARGV[3])
+redis.call("SETEX", KEYS[1], ARGV[2], ARGV[3])
+redis.call("ZADD", KEYS[2], ARGV[1], KEYS[1])
+for i = 4, table.getn(ARGV)-1, 2 do
+	redis.call("HSET", KEYS[3], ARGV[i], ARGV[i+1])
+end
+redis.call("EXPIRE", KEYS[3], ARGV[2])
+redis.call("ZADD", KEYS[4], ARGV[1], KEYS[3])
 return redis.status_reply("OK")`)
 
-// WriteProcessInfo writes process information to redis with expiration
-// set to the value ttl.
-func (r *RDB) WriteProcessInfo(ps *base.ProcessInfo, ttl time.Duration) error {
-	bytes, err := json.Marshal(ps)
+// WriteProcessState writes process state data to redis with expiration  set to the value ttl.
+func (r *RDB) WriteProcessState(ps *base.ProcessState, ttl time.Duration) error {
+	info := ps.Get()
+	bytes, err := json.Marshal(info)
 	if err != nil {
 		return err
 	}
-	// Note: Add key to ZSET with expiration time as score.
-	// ref: https://github.com/antirez/redis/issues/135#issuecomment-2361996
+	var args []interface{} // args to the lua script
 	exp := time.Now().Add(ttl).UTC()
-	key := base.ProcessInfoKey(ps.Host, ps.PID)
-	return writeProcessInfoCmd.Run(r.client, []string{base.AllProcesses, key}, float64(exp.Unix()), ttl.Seconds(), string(bytes)).Err()
-}
-
-// ReadProcessInfo reads process information stored in redis.
-func (r *RDB) ReadProcessInfo(host string, pid int) (*base.ProcessInfo, error) {
-	key := base.ProcessInfoKey(host, pid)
-	data, err := r.client.Get(key).Result()
-	if err != nil {
-		return nil, err
-	}
-	var pinfo base.ProcessInfo
-	err = json.Unmarshal([]byte(data), &pinfo)
-	if err != nil {
-		return nil, err
+	workers := ps.GetWorkers()
+	args = append(args, float64(exp.Unix()), ttl.Seconds(), bytes)
+	for _, w := range workers {
+		bytes, err := json.Marshal(w)
+		if err != nil {
+			continue // skip bad data
+		}
+		args = append(args, w.ID.String(), bytes)
 	}
-	return &pinfo, nil
+	pkey := base.ProcessInfoKey(info.Host, info.PID)
+	wkey := base.WorkersKey(info.Host, info.PID)
+	return writeProcessInfoCmd.Run(r.client,
+		[]string{pkey, base.AllProcesses, wkey, base.AllWorkers},
+		args...).Err()
 }
 
 // KEYS[1] -> asynq:ps
 // KEYS[2] -> asynq:ps:<host:pid>
+// KEYS[3] -> asynq:workers
+// KEYS[4] -> asynq:workers<host:pid>
 var clearProcessInfoCmd = redis.NewScript(`
 redis.call("ZREM", KEYS[1], KEYS[2])
 redis.call("DEL", KEYS[2])
+redis.call("ZREM", KEYS[3], KEYS[4])
+redis.call("DEL", KEYS[4])
 return redis.status_reply("OK")`)
 
-// ClearProcessInfo deletes process information from redis.
-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()
+// ClearProcessState deletes process state data from redis.
+func (r *RDB) ClearProcessState(ps *base.ProcessState) error {
+	info := ps.Get()
+	host, pid := info.Host, info.PID
+	pkey := base.ProcessInfoKey(host, pid)
+	wkey := base.WorkersKey(host, pid)
+	return clearProcessInfoCmd.Run(r.client,
+		[]string{base.AllProcesses, pkey, base.AllWorkers, wkey}).Err()
 }
 
 // CancelationPubSub returns a pubsub for cancelation messages.
diff --git a/processor.go b/processor.go
index edf8de6..0852461 100644
--- a/processor.go
+++ b/processor.go
@@ -20,6 +20,8 @@ import (
 type processor struct {
 	rdb *rdb.RDB
 
+	ps *base.ProcessState
+
 	handler Handler
 
 	queueConfig map[string]int
@@ -32,9 +34,6 @@ type processor struct {
 	// channel via which to send sync requests to syncer.
 	syncRequestCh chan<- *syncRequest
 
-	// channel to send worker count updates.
-	workerCh chan<- int
-
 	// rate limiter to prevent spamming logs with a bunch of errors.
 	errLogLimiter *rate.Limiter
 
@@ -60,23 +59,23 @@ type processor struct {
 type retryDelayFunc func(n int, err error, task *Task) time.Duration
 
 // newProcessor constructs a new processor.
-func newProcessor(r *rdb.RDB, queues map[string]int, strict bool, concurrency int, fn retryDelayFunc,
-	syncRequestCh chan<- *syncRequest, workerCh chan<- int, cancelations *base.Cancelations) *processor {
-	qcfg := normalizeQueueCfg(queues)
+func newProcessor(r *rdb.RDB, ps *base.ProcessState, fn retryDelayFunc, syncCh chan<- *syncRequest, c *base.Cancelations) *processor {
+	info := ps.Get()
+	qcfg := normalizeQueueCfg(info.Queues)
 	orderedQueues := []string(nil)
-	if strict {
+	if info.StrictPriority {
 		orderedQueues = sortByPriority(qcfg)
 	}
 	return &processor{
 		rdb:            r,
+		ps:             ps,
 		queueConfig:    qcfg,
 		orderedQueues:  orderedQueues,
 		retryDelayFunc: fn,
-		syncRequestCh:  syncRequestCh,
-		workerCh:       workerCh,
-		cancelations:   cancelations,
+		syncRequestCh:  syncCh,
+		cancelations:   c,
 		errLogLimiter:  rate.NewLimiter(rate.Every(3*time.Second), 1),
-		sema:           make(chan struct{}, concurrency),
+		sema:           make(chan struct{}, info.Concurrency),
 		done:           make(chan struct{}),
 		abort:          make(chan struct{}),
 		quit:           make(chan struct{}),
@@ -166,10 +165,10 @@ func (p *processor) exec() {
 		p.requeue(msg)
 		return
 	case p.sema <- struct{}{}: // acquire token
-		p.workerCh <- 1
+		p.ps.AddWorkerStats(msg, time.Now())
 		go func() {
 			defer func() {
-				p.workerCh <- -1
+				p.ps.DeleteWorkerStats(msg)
 				<-p.sema /* release token */
 			}()
 
diff --git a/subscriber.go b/subscriber.go
index fc50c06..e8d3731 100644
--- a/subscriber.go
+++ b/subscriber.go
@@ -52,8 +52,8 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 				logger.info("Subscriber done")
 				return
 			case msg := <-cancelCh:
-				cancel := s.cancelations.Get(msg.Payload)
-				if cancel != nil {
+				cancel, ok := s.cancelations.Get(msg.Payload)
+				if ok {
 					cancel()
 				}
 			}
diff --git a/tools/asynqmon/cmd/ps.go b/tools/asynqmon/cmd/ps.go
index 7761886..c4c3867 100644
--- a/tools/asynqmon/cmd/ps.go
+++ b/tools/asynqmon/cmd/ps.go
@@ -73,7 +73,7 @@ func ps(cmd *cobra.Command, args []string) {
 	printRows := func(w io.Writer, tmpl string) {
 		for _, ps := range processes {
 			fmt.Fprintf(w, tmpl,
-				ps.Host, ps.PID, ps.State,
+				ps.Host, ps.PID, ps.Status,
 				fmt.Sprintf("%d/%d", ps.ActiveWorkerCount, ps.Concurrency),
 				formatQueues(ps.Queues), timeAgo(ps.Started))
 		}


Test patch:
diff --git a/heartbeat_test.go b/heartbeat_test.go
index ef5defc..c7b1089 100644
--- a/heartbeat_test.go
+++ b/heartbeat_test.go
@@ -35,9 +35,8 @@ func TestHeartbeater(t *testing.T) {
 	for _, tc := range tests {
 		h.FlushDB(t, r)
 
-		stateCh := make(chan string)
-		workerCh := make(chan int)
-		hb := newHeartbeater(rdbClient, tc.host, tc.pid, tc.concurrency, tc.queues, false, tc.interval, stateCh, workerCh)
+		state := base.NewProcessState(tc.host, tc.pid, tc.concurrency, tc.queues, false)
+		hb := newHeartbeater(rdbClient, state, tc.interval)
 
 		var wg sync.WaitGroup
 		hb.start(&wg)
@@ -48,7 +47,7 @@ func TestHeartbeater(t *testing.T) {
 			Queues:      tc.queues,
 			Concurrency: tc.concurrency,
 			Started:     time.Now(),
-			State:       "running",
+			Status:      "running",
 		}
 
 		// allow for heartbeater to write to redis
@@ -73,13 +72,13 @@ func TestHeartbeater(t *testing.T) {
 			continue
 		}
 
-		// state change
-		stateCh <- "stopped"
+		// status change
+		state.SetStatus(base.StatusStopped)
 
 		// allow for heartbeater to write to redis
 		time.Sleep(tc.interval * 2)
 
-		want.State = "stopped"
+		want.Status = "stopped"
 		ps, err = rdbClient.ListProcesses()
 		if err != nil {
 			t.Errorf("could not read process status from redis: %v", err)
diff --git a/internal/base/base_test.go b/internal/base/base_test.go
index 225dcc6..ee43185 100644
--- a/internal/base/base_test.go
+++ b/internal/base/base_test.go
@@ -5,8 +5,14 @@
 package base
 
 import (
+	"context"
+	"math/rand"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/rs/xid"
 )
 
 func TestQueueKey(t *testing.T) {
@@ -74,7 +80,137 @@ func TestProcessInfoKey(t *testing.T) {
 	for _, tc := range tests {
 		got := ProcessInfoKey(tc.hostname, tc.pid)
 		if got != tc.want {
-			t.Errorf("ProcessInfoKey(%s, %d) = %s, want %s", tc.hostname, tc.pid, got, tc.want)
+			t.Errorf("ProcessInfoKey(%q, %d) = %q, want %q", tc.hostname, tc.pid, got, tc.want)
 		}
 	}
 }
+
+func TestWorkersKey(t *testing.T) {
+	tests := []struct {
+		hostname string
+		pid      int
+		want     string
+	}{
+		{"localhost", 9876, "asynq:workers:localhost:9876"},
+		{"127.0.0.1", 1234, "asynq:workers:127.0.0.1:1234"},
+	}
+
+	for _, tc := range tests {
+		got := WorkersKey(tc.hostname, tc.pid)
+		if got != tc.want {
+			t.Errorf("WorkersKey(%q, %d) = %q, want = %q", tc.hostname, tc.pid, got, tc.want)
+		}
+	}
+}
+
+// Test for process state being accessed by multiple goroutines.
+// Run with -race flag to check for data race.
+func TestProcessStateConcurrentAccess(t *testing.T) {
+	ps := NewProcessState("127.0.0.1", 1234, 10, map[string]int{"default": 1}, false)
+	var wg sync.WaitGroup
+	started := time.Now()
+	msgs := []*TaskMessage{
+		&TaskMessage{ID: xid.New(), Type: "type1", Payload: map[string]interface{}{"user_id": 42}},
+		&TaskMessage{ID: xid.New(), Type: "type2"},
+		&TaskMessage{ID: xid.New(), Type: "type3"},
+	}
+
+	// Simulate hearbeater calling SetStatus and SetStarted.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ps.SetStarted(started)
+		ps.SetStatus(StatusRunning)
+	}()
+
+	// Simulate processor starting worker goroutines.
+	for _, msg := range msgs {
+		wg.Add(1)
+		ps.AddWorkerStats(msg, time.Now())
+		go func(msg *TaskMessage) {
+			defer wg.Done()
+			time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond)
+			ps.DeleteWorkerStats(msg)
+		}(msg)
+	}
+
+	// Simulate hearbeater calling Get and GetWorkers
+	wg.Add(1)
+	go func() {
+		wg.Done()
+		for i := 0; i < 5; i++ {
+			ps.Get()
+			ps.GetWorkers()
+			time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
+		}
+	}()
+
+	wg.Wait()
+
+	want := &ProcessInfo{
+		Host:              "127.0.0.1",
+		PID:               1234,
+		Concurrency:       10,
+		Queues:            map[string]int{"default": 1},
+		StrictPriority:    false,
+		Status:            "running",
+		Started:           started,
+		ActiveWorkerCount: 0,
+	}
+
+	got := ps.Get()
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("(*ProcessState).Get() = %+v, want %+v; (-want,+got)\n%s",
+			got, want, diff)
+	}
+}
+
+// Test for cancelations being accessed by multiple goroutines.
+// Run with -race flag to check for data race.
+func TestCancelationsConcurrentAccess(t *testing.T) {
+	c := NewCancelations()
+
+	_, cancel1 := context.WithCancel(context.Background())
+	_, cancel2 := context.WithCancel(context.Background())
+	_, cancel3 := context.WithCancel(context.Background())
+	var key1, key2, key3 = "key1", "key2", "key3"
+
+	var wg sync.WaitGroup
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		c.Add(key1, cancel1)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		c.Add(key2, cancel2)
+		time.Sleep(200 * time.Millisecond)
+		c.Delete(key2)
+	}()
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		c.Add(key3, cancel3)
+	}()
+
+	wg.Wait()
+
+	_, ok := c.Get(key1)
+	if !ok {
+		t.Errorf("(*Cancelations).Get(%q) = _, false, want <function>, true", key1)
+	}
+
+	_, ok = c.Get(key2)
+	if ok {
+		t.Errorf("(*Cancelations).Get(%q) = _, true, want <nil>, false", key2)
+	}
+
+	funcs := c.GetAll()
+	if len(funcs) != 2 {
+		t.Errorf("(*Cancelations).GetAll() returns %d functions, want 2", len(funcs))
+	}
+}
diff --git a/internal/rdb/inspect_test.go b/internal/rdb/inspect_test.go
index cd671c2..48bcb56 100644
--- a/internal/rdb/inspect_test.go
+++ b/internal/rdb/inspect_test.go
@@ -2054,32 +2054,51 @@ func TestRemoveQueueError(t *testing.T) {
 func TestListProcesses(t *testing.T) {
 	r := setup(t)
 
-	ps1 := &base.ProcessInfo{
+	started1 := time.Now().Add(-time.Hour)
+	ps1 := base.NewProcessState("do.droplet1", 1234, 10, map[string]int{"default": 1}, false)
+	ps1.SetStarted(started1)
+	ps1.SetStatus(base.StatusRunning)
+	info1 := &base.ProcessInfo{
 		Concurrency:       10,
 		Queues:            map[string]int{"default": 1},
 		Host:              "do.droplet1",
 		PID:               1234,
-		State:             "running",
-		Started:           time.Now().Add(-time.Hour),
-		ActiveWorkerCount: 5,
+		Status:            "running",
+		Started:           started1,
+		ActiveWorkerCount: 0,
 	}
 
-	ps2 := &base.ProcessInfo{
+	started2 := time.Now().Add(-2 * time.Hour)
+	ps2 := base.NewProcessState("do.droplet2", 9876, 20, map[string]int{"email": 1}, false)
+	ps2.SetStarted(started2)
+	ps2.SetStatus(base.StatusStopped)
+	ps2.AddWorkerStats(h.NewTaskMessage("send_email", nil), time.Now())
+	info2 := &base.ProcessInfo{
 		Concurrency:       20,
 		Queues:            map[string]int{"email": 1},
 		Host:              "do.droplet2",
 		PID:               9876,
-		State:             "stopped",
-		Started:           time.Now().Add(-2 * time.Hour),
-		ActiveWorkerCount: 20,
+		Status:            "stopped",
+		Started:           started2,
+		ActiveWorkerCount: 1,
 	}
 
 	tests := []struct {
-		processes []*base.ProcessInfo
+		processes []*base.ProcessState
+		want      []*base.ProcessInfo
 	}{
-		{processes: []*base.ProcessInfo{}},
-		{processes: []*base.ProcessInfo{ps1}},
-		{processes: []*base.ProcessInfo{ps1, ps2}},
+		{
+			processes: []*base.ProcessState{},
+			want:      []*base.ProcessInfo{},
+		},
+		{
+			processes: []*base.ProcessState{ps1},
+			want:      []*base.ProcessInfo{info1},
+		},
+		{
+			processes: []*base.ProcessState{ps1, ps2},
+			want:      []*base.ProcessInfo{info1, info2},
+		},
 	}
 
 	ignoreOpt := cmpopts.IgnoreUnexported(base.ProcessInfo{})
@@ -2088,7 +2107,7 @@ func TestListProcesses(t *testing.T) {
 		h.FlushDB(t, r.client)
 
 		for _, ps := range tc.processes {
-			if err := r.WriteProcessInfo(ps, 5*time.Second); err != nil {
+			if err := r.WriteProcessState(ps, 5*time.Second); err != nil {
 				t.Fatal(err)
 			}
 		}
@@ -2097,7 +2116,7 @@ func TestListProcesses(t *testing.T) {
 		if err != nil {
 			t.Errorf("r.ListProcesses returned an error: %v", err)
 		}
-		if diff := cmp.Diff(tc.processes, got, h.SortProcessInfoOpt, ignoreOpt); diff != "" {
+		if diff := cmp.Diff(tc.want, got, h.SortProcessInfoOpt, ignoreOpt); diff != "" {
 			t.Errorf("r.ListProcesses returned %v, want %v; (-want,+got)\n%s",
 				got, tc.processes, diff)
 		}
diff --git a/internal/rdb/rdb_test.go b/internal/rdb/rdb_test.go
index accc52d..e0b4669 100644
--- a/internal/rdb/rdb_test.go
+++ b/internal/rdb/rdb_test.go
@@ -5,8 +5,8 @@
 package rdb
 
 import (
+	"encoding/json"
 	"fmt"
-	"strconv"
 	"testing"
 	"time"
 
@@ -741,81 +741,237 @@ func TestCheckAndEnqueue(t *testing.T) {
 	}
 }
 
-func TestReadWriteClearProcessInfo(t *testing.T) {
+func TestWriteProcessState(t *testing.T) {
 	r := setup(t)
-	pinfo := &base.ProcessInfo{
+	host, pid := "localhost", 98765
+	queues := map[string]int{"default": 2, "email": 5, "low": 1}
+
+	started := time.Now()
+	ps := base.NewProcessState(host, pid, 10, queues, false)
+	ps.SetStarted(started)
+	ps.SetStatus(base.StatusRunning)
+	ttl := 5 * time.Second
+
+	h.FlushDB(t, r.client)
+
+	err := r.WriteProcessState(ps, ttl)
+	if err != nil {
+		t.Errorf("r.WriteProcessState returned an error: %v", err)
+	}
+
+	// Check ProcessInfo was written correctly
+	pkey := base.ProcessInfoKey(host, pid)
+	data := r.client.Get(pkey).Val()
+	var got base.ProcessInfo
+	err = json.Unmarshal([]byte(data), &got)
+	if err != nil {
+		t.Fatalf("could not decode json: %v", err)
+	}
+	want := base.ProcessInfo{
+		Host:              "localhost",
+		PID:               98765,
 		Concurrency:       10,
 		Queues:            map[string]int{"default": 2, "email": 5, "low": 1},
-		PID:               98765,
-		Host:              "localhost",
-		State:             "running",
-		Started:           time.Now(),
-		ActiveWorkerCount: 1,
+		StrictPriority:    false,
+		Status:            "running",
+		Started:           started,
+		ActiveWorkerCount: 0,
+	}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("persisted ProcessInfo was %v, want %v; (-want,+got)\n%s",
+			got, want, diff)
+	}
+	// Check ProcessInfo TTL was set correctly
+	gotTTL := r.client.TTL(pkey).Val()
+	timeCmpOpt := cmpopts.EquateApproxTime(time.Second)
+	if !cmp.Equal(ttl, gotTTL, timeCmpOpt) {
+		t.Errorf("TTL of %q was %v, want %v", pkey, gotTTL, ttl)
+	}
+	// Check ProcessInfo key was added to the set correctly
+	gotProcesses := r.client.ZRange(base.AllProcesses, 0, -1).Val()
+	wantProcesses := []string{pkey}
+	if diff := cmp.Diff(wantProcesses, gotProcesses); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllProcesses, gotProcesses, wantProcesses)
 	}
 
-	tests := []struct {
-		pi  *base.ProcessInfo
-		ttl time.Duration
-	}{
-		{pinfo, 5 * time.Second},
+	// Check WorkersInfo was written correctly
+	wkey := base.WorkersKey(host, pid)
+	workerExist := r.client.Exists(wkey).Val()
+	if workerExist != 0 {
+		t.Errorf("%q key exists", wkey)
+	}
+	// Check WorkersInfo key was added to the set correctly
+	gotWorkerKeys := r.client.ZRange(base.AllWorkers, 0, -1).Val()
+	wantWorkerKeys := []string{wkey}
+	if diff := cmp.Diff(wantWorkerKeys, gotWorkerKeys); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllWorkers, gotWorkerKeys, wantWorkerKeys)
 	}
+}
 
-	for _, tc := range tests {
-		h.FlushDB(t, r.client)
+func TestWriteProcessStateWithWorkers(t *testing.T) {
+	r := setup(t)
+	host, pid := "localhost", 98765
+	queues := map[string]int{"default": 2, "email": 5, "low": 1}
+	concurrency := 10
+
+	started := time.Now().Add(-10 * time.Minute)
+	w1Started := time.Now().Add(-time.Minute)
+	w2Started := time.Now().Add(-time.Second)
+	msg1 := h.NewTaskMessage("send_email", map[string]interface{}{"user_id": "123"})
+	msg2 := h.NewTaskMessage("gen_thumbnail", map[string]interface{}{"path": "some/path/to/imgfile"})
+	ps := base.NewProcessState(host, pid, concurrency, queues, false)
+	ps.SetStarted(started)
+	ps.SetStatus(base.StatusRunning)
+	ps.AddWorkerStats(msg1, w1Started)
+	ps.AddWorkerStats(msg2, w2Started)
+	ttl := 5 * time.Second
 
-		err := r.WriteProcessInfo(tc.pi, tc.ttl)
-		if err != nil {
-			t.Errorf("r.WriteProcessInfo returned an error: %v", err)
-			continue
-		}
+	h.FlushDB(t, r.client)
 
-		got, err := r.ReadProcessInfo(tc.pi.Host, tc.pi.PID)
-		if err != nil {
-			t.Errorf("r.ReadProcessInfo returned an error: %v", err)
-			continue
-		}
+	err := r.WriteProcessState(ps, ttl)
+	if err != nil {
+		t.Errorf("r.WriteProcessState returned an error: %v", err)
+	}
 
-		ignoreOpt := cmpopts.IgnoreUnexported(base.ProcessInfo{})
-		if diff := cmp.Diff(tc.pi, got, ignoreOpt); diff != "" {
-			t.Errorf("r.ReadProcessInfo(%q, %d) = %+v, want %+v; (-want,+got)\n%s",
-				tc.pi.Host, tc.pi.PID, got, tc.pi, diff)
-		}
+	// Check ProcessInfo was written correctly
+	pkey := base.ProcessInfoKey(host, pid)
+	data := r.client.Get(pkey).Val()
+	var got base.ProcessInfo
+	err = json.Unmarshal([]byte(data), &got)
+	if err != nil {
+		t.Fatalf("could not decode json: %v", err)
+	}
+	want := base.ProcessInfo{
+		Host:              host,
+		PID:               pid,
+		Concurrency:       concurrency,
+		Queues:            queues,
+		StrictPriority:    false,
+		Status:            "running",
+		Started:           started,
+		ActiveWorkerCount: 2,
+	}
+	if diff := cmp.Diff(want, got); diff != "" {
+		t.Errorf("persisted ProcessInfo was %v, want %v; (-want,+got)\n%s",
+			got, want, diff)
+	}
+	// Check ProcessInfo TTL was set correctly
+	gotTTL := r.client.TTL(pkey).Val()
+	timeCmpOpt := cmpopts.EquateApproxTime(time.Second)
+	if !cmp.Equal(ttl, gotTTL, timeCmpOpt) {
+		t.Errorf("TTL of %q was %v, want %v", pkey, gotTTL, ttl)
+	}
+	// Check ProcessInfo key was added to the set correctly
+	gotProcesses := r.client.ZRange(base.AllProcesses, 0, -1).Val()
+	wantProcesses := []string{pkey}
+	if diff := cmp.Diff(wantProcesses, gotProcesses); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllProcesses, gotProcesses, wantProcesses)
+	}
 
-		key := base.ProcessInfoKey(tc.pi.Host, tc.pi.PID)
-		gotTTL := r.client.TTL(key).Val()
-		if !cmp.Equal(tc.ttl, gotTTL, timeCmpOpt) {
-			t.Errorf("redis TTL %q returned %v, want %v", key, gotTTL, tc.ttl)
+	// Check WorkersInfo was written correctly
+	wkey := base.WorkersKey(host, pid)
+	wdata := r.client.HGetAll(wkey).Val()
+	if len(wdata) != 2 {
+		t.Fatalf("HGETALL %q returned a hash of size %d, want 2", wkey, len(wdata))
+	}
+	gotWorkers := make(map[string]*base.WorkerInfo)
+	for key, val := range wdata {
+		var w base.WorkerInfo
+		if err := json.Unmarshal([]byte(val), &w); err != nil {
+			t.Fatalf("could not unmarshal worker's data: %v", err)
 		}
+		gotWorkers[key] = &w
+	}
+	wantWorkers := map[string]*base.WorkerInfo{
+		msg1.ID.String(): &base.WorkerInfo{
+			Host:    host,
+			PID:     pid,
+			ID:      msg1.ID,
+			Type:    msg1.Type,
+			Queue:   msg1.Queue,
+			Payload: msg1.Payload,
+			Started: w1Started,
+		},
+		msg2.ID.String(): &base.WorkerInfo{
+			Host:    host,
+			PID:     pid,
+			ID:      msg2.ID,
+			Type:    msg2.Type,
+			Queue:   msg2.Queue,
+			Payload: msg2.Payload,
+			Started: w2Started,
+		},
+	}
+	if diff := cmp.Diff(wantWorkers, gotWorkers); diff != "" {
+		t.Errorf("persisted workers info was %v, want %v; (-want,+got)\n%s",
+			gotWorkers, wantWorkers, diff)
+	}
 
-		now := time.Now().UTC()
-		allKeys, err := r.client.ZRangeByScore(base.AllProcesses, &redis.ZRangeBy{
-			Min: strconv.Itoa(int(now.Unix())),
-			Max: "+inf",
-		}).Result()
-		if err != nil {
-			t.Errorf("redis ZRANGEBYSCORE %q %d +inf returned an error: %v",
-				base.AllProcesses, now.Unix(), err)
-			continue
-		}
+	// Check WorkersInfo TTL was set correctly
+	gotTTL = r.client.TTL(wkey).Val()
+	if !cmp.Equal(ttl, gotTTL, timeCmpOpt) {
+		t.Errorf("TTL of %q was %v, want %v", wkey, gotTTL, ttl)
+	}
+	// Check WorkersInfo key was added to the set correctly
+	gotWorkerKeys := r.client.ZRange(base.AllWorkers, 0, -1).Val()
+	wantWorkerKeys := []string{wkey}
+	if diff := cmp.Diff(wantWorkerKeys, gotWorkerKeys); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllWorkers, gotWorkerKeys, wantWorkerKeys)
+	}
+}
 
-		wantAllKeys := []string{key}
-		if diff := cmp.Diff(wantAllKeys, allKeys); diff != "" {
-			t.Errorf("all keys = %v, want %v; (-want,+got)\n%s", allKeys, wantAllKeys, diff)
-		}
+func TestClearProcessState(t *testing.T) {
+	r := setup(t)
+	host, pid := "127.0.0.1", 1234
 
-		if err := r.ClearProcessInfo(tc.pi); err != nil {
-			t.Errorf("r.ClearProcessInfo returned an error: %v", err)
-			continue
-		}
+	h.FlushDB(t, r.client)
 
-		// 1 means key exists
-		if r.client.Exists(key).Val() == 1 {
-			t.Errorf("expected %q to be deleted", key)
-		}
+	pkey := base.ProcessInfoKey(host, pid)
+	wkey := base.WorkersKey(host, pid)
+	otherPKey := base.ProcessInfoKey("otherhost", 12345)
+	otherWKey := base.WorkersKey("otherhost", 12345)
+	// Populate the keys.
+	if err := r.client.Set(pkey, "process-info", 0).Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.client.HSet(wkey, "worker-key", "worker-info").Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.client.ZAdd(base.AllProcesses, &redis.Z{Member: pkey}).Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.client.ZAdd(base.AllProcesses, &redis.Z{Member: otherPKey}).Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.client.ZAdd(base.AllWorkers, &redis.Z{Member: wkey}).Err(); err != nil {
+		t.Fatal(err)
+	}
+	if err := r.client.ZAdd(base.AllWorkers, &redis.Z{Member: otherWKey}).Err(); err != nil {
+		t.Fatal(err)
+	}
 
-		if r.client.ZCard(base.AllProcesses).Val() != 0 {
-			t.Errorf("expected %q to be empty", base.AllProcesses)
-		}
+	ps := base.NewProcessState(host, pid, 10, map[string]int{"default": 1}, false)
+
+	err := r.ClearProcessState(ps)
+	if err != nil {
+		t.Fatalf("(*RDB).ClearProcessState failed: %v", err)
+	}
 
+	// Check all keys are cleared
+	if r.client.Exists(pkey).Val() != 0 {
+		t.Errorf("Redis key %q exists", pkey)
+	}
+	if r.client.Exists(wkey).Val() != 0 {
+		t.Errorf("Redis key %q exists", wkey)
+	}
+	gotProcessKeys := r.client.ZRange(base.AllProcesses, 0, -1).Val()
+	wantProcessKeys := []string{otherPKey}
+	if diff := cmp.Diff(wantProcessKeys, gotProcessKeys); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllProcesses, gotProcessKeys, wantProcessKeys)
+	}
+	gotWorkerKeys := r.client.ZRange(base.AllWorkers, 0, -1).Val()
+	wantWorkerKeys := []string{otherWKey}
+	if diff := cmp.Diff(wantWorkerKeys, gotWorkerKeys); diff != "" {
+		t.Errorf("%q contained %v, want %v", base.AllWorkers, gotWorkerKeys, wantWorkerKeys)
 	}
 }
diff --git a/processor_test.go b/processor_test.go
index 77224c0..177547b 100644
--- a/processor_test.go
+++ b/processor_test.go
@@ -68,8 +68,9 @@ func TestProcessorSuccess(t *testing.T) {
 		}
 		workerCh := make(chan int)
 		go fakeHeartbeater(workerCh)
+		ps := base.NewProcessState("localhost", 1234, 10, defaultQueueConfig, false)
 		cancelations := base.NewCancelations()
-		p := newProcessor(rdbClient, defaultQueueConfig, false, 10, defaultDelayFunc, nil, workerCh, cancelations)
+		p := newProcessor(rdbClient, ps, defaultDelayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		var wg sync.WaitGroup
@@ -156,8 +157,9 @@ func TestProcessorRetry(t *testing.T) {
 		}
 		workerCh := make(chan int)
 		go fakeHeartbeater(workerCh)
+		ps := base.NewProcessState("localhost", 1234, 10, defaultQueueConfig, false)
 		cancelations := base.NewCancelations()
-		p := newProcessor(rdbClient, defaultQueueConfig, false, 10, delayFunc, nil, workerCh, cancelations)
+		p := newProcessor(rdbClient, ps, delayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		var wg sync.WaitGroup
@@ -219,7 +221,8 @@ func TestProcessorQueues(t *testing.T) {
 
 	for _, tc := range tests {
 		cancelations := base.NewCancelations()
-		p := newProcessor(nil, tc.queueCfg, false, 10, defaultDelayFunc, nil, nil, cancelations)
+		ps := base.NewProcessState("localhost", 1234, 10, tc.queueCfg, false)
+		p := newProcessor(nil, ps, 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",
@@ -288,8 +291,8 @@ func TestProcessorWithStrictPriority(t *testing.T) {
 		workerCh := make(chan int)
 		go fakeHeartbeater(workerCh)
 		cancelations := base.NewCancelations()
-		p := newProcessor(rdbClient, queueCfg, true /* strict */, 1, /* concurrency */
-			defaultDelayFunc, nil, workerCh, cancelations)
+		ps := base.NewProcessState("localhost", 1234, 1 /* concurrency */, queueCfg, true /*strict*/)
+		p := newProcessor(rdbClient, ps, defaultDelayFunc, nil, cancelations)
 		p.handler = HandlerFunc(handler)
 
 		var wg sync.WaitGroup