mirror of
https://github.com/hibiken/asynq.git
synced 2024-12-25 23:32:17 +08:00
Do not start worker goroutine for task already exceeded its deadline
This commit is contained in:
parent
007fac8055
commit
7382e2aeb8
@ -34,6 +34,7 @@ A system can consist of multiple worker servers and brokers, giving way to high
|
|||||||
- Scheduling of tasks
|
- Scheduling of tasks
|
||||||
- Durability since tasks are written to Redis
|
- Durability since tasks are written to Redis
|
||||||
- [Retries](https://github.com/hibiken/asynq/wiki/Task-Retry) of failed tasks
|
- [Retries](https://github.com/hibiken/asynq/wiki/Task-Retry) of failed tasks
|
||||||
|
- Automatic recovery of tasks in the event of a worker crash
|
||||||
- [Weighted priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#weighted-priority-queues)
|
- [Weighted priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#weighted-priority-queues)
|
||||||
- [Strict priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#strict-priority-queues)
|
- [Strict priority queues](https://github.com/hibiken/asynq/wiki/Priority-Queues#strict-priority-queues)
|
||||||
- Low latency to add a task since writes are fast in Redis
|
- Low latency to add a task since writes are fast in Redis
|
||||||
|
26
processor.go
26
processor.go
@ -191,9 +191,19 @@ func (p *processor) exec() {
|
|||||||
p.cancelations.Delete(msg.ID.String())
|
p.cancelations.Delete(msg.ID.String())
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
// check context before starting a worker goroutine.
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
// already canceled (e.g. deadline exceeded).
|
||||||
|
p.retryOrKill(ctx, msg, ctx.Err())
|
||||||
|
return
|
||||||
|
default:
|
||||||
|
}
|
||||||
|
|
||||||
resCh := make(chan error, 1)
|
resCh := make(chan error, 1)
|
||||||
task := NewTask(msg.Type, msg.Payload)
|
go func() {
|
||||||
go func() { resCh <- perform(ctx, task, p.handler) }()
|
resCh <- perform(ctx, NewTask(msg.Type, msg.Payload), p.handler)
|
||||||
|
}()
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case <-p.abort:
|
case <-p.abort:
|
||||||
@ -202,10 +212,6 @@ func (p *processor) exec() {
|
|||||||
p.requeue(msg)
|
p.requeue(msg)
|
||||||
return
|
return
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
p.logger.Debugf("Retrying task. task id=%s", msg.ID) // TODO: Improve this log message and above
|
|
||||||
if p.errHandler != nil {
|
|
||||||
p.errHandler.HandleError(ctx, task, ctx.Err())
|
|
||||||
}
|
|
||||||
p.retryOrKill(ctx, msg, ctx.Err())
|
p.retryOrKill(ctx, msg, ctx.Err())
|
||||||
return
|
return
|
||||||
case resErr := <-resCh:
|
case resErr := <-resCh:
|
||||||
@ -214,9 +220,6 @@ func (p *processor) exec() {
|
|||||||
// 2) Retry -> Removes the message from InProgress & Adds the message to Retry
|
// 2) Retry -> Removes the message from InProgress & Adds the message to Retry
|
||||||
// 3) Kill -> Removes the message from InProgress & Adds the message to Dead
|
// 3) Kill -> Removes the message from InProgress & Adds the message to Dead
|
||||||
if resErr != nil {
|
if resErr != nil {
|
||||||
if p.errHandler != nil {
|
|
||||||
p.errHandler.HandleError(ctx, task, resErr)
|
|
||||||
}
|
|
||||||
p.retryOrKill(ctx, msg, resErr)
|
p.retryOrKill(ctx, msg, resErr)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -255,7 +258,11 @@ func (p *processor) markAsDone(ctx context.Context, msg *base.TaskMessage) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (p *processor) retryOrKill(ctx context.Context, msg *base.TaskMessage, err error) {
|
func (p *processor) retryOrKill(ctx context.Context, msg *base.TaskMessage, err error) {
|
||||||
|
if p.errHandler != nil {
|
||||||
|
p.errHandler.HandleError(ctx, NewTask(msg.Type, msg.Payload), err)
|
||||||
|
}
|
||||||
if msg.Retried >= msg.Retry {
|
if msg.Retried >= msg.Retry {
|
||||||
|
p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
|
||||||
p.kill(ctx, msg, err)
|
p.kill(ctx, msg, err)
|
||||||
} else {
|
} else {
|
||||||
p.retry(ctx, msg, err)
|
p.retry(ctx, msg, err)
|
||||||
@ -284,7 +291,6 @@ func (p *processor) retry(ctx context.Context, msg *base.TaskMessage, e error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (p *processor) kill(ctx context.Context, msg *base.TaskMessage, e error) {
|
func (p *processor) kill(ctx context.Context, msg *base.TaskMessage, e error) {
|
||||||
p.logger.Warnf("Retry exhausted for task id=%s", msg.ID)
|
|
||||||
err := p.broker.Kill(msg, e.Error())
|
err := p.broker.Kill(msg, e.Error())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.DeadQueue)
|
errMsg := fmt.Sprintf("Could not move task id=%s from %q to %q", msg.ID, base.InProgressQueue, base.DeadQueue)
|
||||||
|
Loading…
Reference in New Issue
Block a user