diff --git a/processor.go b/processor.go index 71770cb..a50e474 100644 --- a/processor.go +++ b/processor.go @@ -59,11 +59,10 @@ func (p *processor) start() { // exec pulls a task out of the queue and starts a worker goroutine to // process the task. 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. const timeout = 5 * time.Second - // TODO(hibiken): sort the list of queues in order of priority - msg, err := p.rdb.dequeue(timeout, p.rdb.listQueues()...) + msg, err := p.rdb.dequeue(defaultQueue, timeout) if err != nil { switch err { case errQueuePopTimeout: @@ -82,8 +81,8 @@ func (p *processor) exec() { p.sema <- struct{}{} // acquire token go func(task *Task) { defer func() { - if err := p.rdb.srem(inProgress, msg); err != nil { - log.Printf("[SERVER ERROR] SREM failed: %v\n", err) + if err := p.rdb.lrem(inProgress, msg); err != nil { + log.Printf("[SERVER ERROR] LREM failed: %v\n", err) } <-p.sema // release token }() diff --git a/rdb.go b/rdb.go index d394c82..ee50e43 100644 --- a/rdb.go +++ b/rdb.go @@ -13,13 +13,13 @@ import ( // Redis keys const ( - queuePrefix = "asynq:queues:" // LIST - asynq:queues: - allQueues = "asynq:queues" // SET - scheduled = "asynq:scheduled" // ZSET - retry = "asynq:retry" // ZSET - dead = "asynq:dead" // ZSET - inProgress = "asynq:in_progress" // SET - heartbeatPrefix = "asynq:heartbeat:" // STRING - asynq:heartbeat: + queuePrefix = "asynq:queues:" // LIST - asynq:queues: + defaultQueue = queuePrefix + "default" // LIST + allQueues = "asynq:queues" // SET + scheduled = "asynq:scheduled" // ZSET + retry = "asynq:retry" // ZSET + dead = "asynq:dead" // ZSET + inProgress = "asynq:in_progress" // SET ) var ( @@ -58,38 +58,33 @@ func (r *rdb) push(msg *taskMessage) error { } // 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. -func (r *rdb) dequeue(timeout time.Duration, keys ...string) (*taskMessage, error) { - // TODO(hibiken): Make BRPOP & SADD atomic. - res, err := r.client.BRPop(timeout, keys...).Result() +// once available, it adds the task to "in progress" list and returns the task. +func (r *rdb) dequeue(qname string, timeout time.Duration) (*taskMessage, error) { + data, err := r.client.BRPopLPush(qname, inProgress, timeout).Result() if err != 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 } - 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 err = json.Unmarshal([]byte(data), &msg) if err != nil { 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 } -func (r *rdb) srem(key string, msg *taskMessage) error { +func (r *rdb) lrem(key string, msg *taskMessage) error { bytes, err := json.Marshal(msg) if err != nil { 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 { - 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 } diff --git a/rdb_test.go b/rdb_test.go index c78d69f..fd5410e 100644 --- a/rdb_test.go +++ b/rdb_test.go @@ -63,7 +63,7 @@ func TestDequeueImmediateReturn(t *testing.T) { } r.push(msg) - res, err := r.dequeue(time.Second, "asynq:queues:csv") + res, err := r.dequeue("asynq:queues:csv", time.Second) if err != nil { t.Fatalf("r.bpop() failed: %v", err) } @@ -71,7 +71,7 @@ func TestDequeueImmediateReturn(t *testing.T) { if !cmp.Equal(res, msg) { 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 { t.Fatalf("len(jobs) = %d, want %d", len(jobs), 1) } @@ -87,7 +87,7 @@ func TestDequeueImmediateReturn(t *testing.T) { func TestDequeueTimeout(t *testing.T) { r := setup() - _, err := r.dequeue(time.Second, "asynq:queues:default") + _, err := r.dequeue("asynq:queues:default", time.Second) if err != errQueuePopTimeout { t.Errorf("err = %v, want %v", err, errQueuePopTimeout) }