go-pkg/asynq
2
0
Fork 0
mirror of https://github.com/hibiken/asynq.git synced 2025-02-23 12:20:19 +08:00
Code Issues Projects Releases Wiki Activity

Use command BRPUSHLPOP to move from queue to in_progress in redis

Browse Source
This commit is contained in:
Ken Hibino 2019-11-23 08:43:41 -08:00
parent 76ceb282a9
commit 4a327933bd
3 changed files with 23 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
processor.go
View File

@ -59,11 +59,10 @@ func (p *processor) start() {
// exec pulls a task out of the queue and starts a worker goroutine to // exec pulls a task out of the queue and starts a worker goroutine to
// process the task. // process the task.
func (p *processor) exec() { func (p *processor) exec() {
// NOTE: BLPOP needs to timeout to avoid blocking forever // NOTE: dequeue needs to timeout to avoid blocking forever
// in case of a program shutdown or additon of a new queue. // in case of a program shutdown or additon of a new queue.
const timeout = 5 * time.Second const timeout = 5 * time.Second
// TODO(hibiken): sort the list of queues in order of priority msg, err := p.rdb.dequeue(defaultQueue, timeout)
msg, err := p.rdb.dequeue(timeout, p.rdb.listQueues()...)
if err != nil { if err != nil {
switch err { switch err {
case errQueuePopTimeout: case errQueuePopTimeout:
@ -82,8 +81,8 @@ func (p *processor) exec() {
p.sema <- struct{}{} // acquire token p.sema <- struct{}{} // acquire token
go func(task *Task) { go func(task *Task) {
defer func() { defer func() {
if err := p.rdb.srem(inProgress, msg); err != nil { if err := p.rdb.lrem(inProgress, msg); err != nil {
log.Printf("[SERVER ERROR] SREM failed: %v\n", err) log.Printf("[SERVER ERROR] LREM failed: %v\n", err)
} }
<-p.sema // release token <-p.sema // release token
}() }()

25
rdb.go
View File

@ -14,12 +14,12 @@ import (
// Redis keys // Redis keys
const ( const (
queuePrefix = "asynq:queues:" // LIST - asynq:queues:<qname> queuePrefix = "asynq:queues:" // LIST - asynq:queues:<qname>
defaultQueue = queuePrefix + "default" // LIST
allQueues = "asynq:queues" // SET allQueues = "asynq:queues" // SET
scheduled = "asynq:scheduled" // ZSET scheduled = "asynq:scheduled" // ZSET
retry = "asynq:retry" // ZSET retry = "asynq:retry" // ZSET
dead = "asynq:dead" // ZSET dead = "asynq:dead" // ZSET
inProgress = "asynq:in_progress" // SET inProgress = "asynq:in_progress" // SET
heartbeatPrefix = "asynq:heartbeat:" // STRING - asynq:heartbeat:<taskID>
) )
var ( var (
@ -58,38 +58,33 @@ func (r *rdb) push(msg *taskMessage) error {
} }
// dequeue blocks until there is a taskMessage available to be processed, // dequeue blocks until there is a taskMessage available to be processed,
// once available, it adds the task to "in progress" set and returns the task. // once available, it adds the task to "in progress" list and returns the task.
func (r *rdb) dequeue(timeout time.Duration, keys ...string) (*taskMessage, error) { func (r *rdb) dequeue(qname string, timeout time.Duration) (*taskMessage, error) {
// TODO(hibiken): Make BRPOP & SADD atomic. data, err := r.client.BRPopLPush(qname, inProgress, timeout).Result()
res, err := r.client.BRPop(timeout, keys...).Result()
if err != nil { if err != nil {
if err != redis.Nil { if err != redis.Nil {
return nil, fmt.Errorf("command BLPOP %v %v failed: %v", timeout, keys, err) return nil, fmt.Errorf("command BRPOPLPUSH %q %q %v failed: %v", qname, inProgress, timeout, err)
} }
return nil, errQueuePopTimeout return nil, errQueuePopTimeout
} }
q, data := res[0], res[1]
err = r.client.SAdd(inProgress, data).Err()
if err != nil {
return nil, fmt.Errorf("command SADD %q %v failed: %v", inProgress, data, err)
}
var msg taskMessage var msg taskMessage
err = json.Unmarshal([]byte(data), &msg) err = json.Unmarshal([]byte(data), &msg)
if err != nil { if err != nil {
return nil, errDeserializeTask return nil, errDeserializeTask
} }
fmt.Printf("[DEBUG] perform task %+v from %s\n", msg, q) fmt.Printf("[DEBUG] perform task %+v from %s\n", msg, qname)
return &msg, nil return &msg, nil
} }
func (r *rdb) srem(key string, msg *taskMessage) error { func (r *rdb) lrem(key string, msg *taskMessage) error {
bytes, err := json.Marshal(msg) bytes, err := json.Marshal(msg)
if err != nil { if err != nil {
return fmt.Errorf("could not encode task into JSON: %v", err) return fmt.Errorf("could not encode task into JSON: %v", err)
} }
err = r.client.SRem(key, string(bytes)).Err() // NOTE: count ZERO means "remove all elements equal to val"
err = r.client.LRem(key, 0, string(bytes)).Err()
if err != nil { if err != nil {
return fmt.Errorf("command SREM %s %s failed: %v", key, string(bytes), err) return fmt.Errorf("command LREM %s 0 %s failed: %v", key, string(bytes), err)
} }
return nil return nil
} }

6
rdb_test.go
View File

@ -63,7 +63,7 @@ func TestDequeueImmediateReturn(t *testing.T) {
} }
r.push(msg) r.push(msg)
res, err := r.dequeue(time.Second, "asynq:queues:csv") res, err := r.dequeue("asynq:queues:csv", time.Second)
if err != nil { if err != nil {
t.Fatalf("r.bpop() failed: %v", err) t.Fatalf("r.bpop() failed: %v", err)
} }
@ -71,7 +71,7 @@ func TestDequeueImmediateReturn(t *testing.T) {
if !cmp.Equal(res, msg) { if !cmp.Equal(res, msg) {
t.Errorf("cmp.Equal(res, msg) = %t, want %t", false, true) t.Errorf("cmp.Equal(res, msg) = %t, want %t", false, true)
} }
jobs := client.SMembers(inProgress).Val() jobs := client.LRange(inProgress, 0, -1).Val()
if len(jobs) != 1 { if len(jobs) != 1 {
t.Fatalf("len(jobs) = %d, want %d", len(jobs), 1) t.Fatalf("len(jobs) = %d, want %d", len(jobs), 1)
} }
@ -87,7 +87,7 @@ func TestDequeueImmediateReturn(t *testing.T) {
func TestDequeueTimeout(t *testing.T) { func TestDequeueTimeout(t *testing.T) {
r := setup() r := setup()
_, err := r.dequeue(time.Second, "asynq:queues:default") _, err := r.dequeue("asynq:queues:default", time.Second)
if err != errQueuePopTimeout { if err != errQueuePopTimeout {
t.Errorf("err = %v, want %v", err, errQueuePopTimeout) t.Errorf("err = %v, want %v", err, errQueuePopTimeout)
} }