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

Print stack track when recovering from a panic in processor

Browse Source
This commit is contained in:
Ken Hibino 2021-01-10 11:20:20 -08:00
parent c5100a9c23
commit 0712e90f23
2 changed files with 14 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
processor.go
View File

@ -8,6 +8,7 @@ import (
"context" "context"
"fmt" "fmt"
"math/rand" "math/rand"
"runtime/debug"
"sort" "sort"
"sync" "sync"
"time" "time"
@ -203,7 +204,7 @@ func (p *processor) exec() {
resCh := make(chan error, 1) resCh := make(chan error, 1)
go func() { go func() {
resCh <- perform(ctx, NewTask(msg.Type, msg.Payload), p.handler) resCh <- p.perform(ctx, NewTask(msg.Type, msg.Payload))
}() }()
select { select {
@ -340,13 +341,14 @@ func (p *processor) queues() []string {
// perform calls the handler with the given task. // perform calls the handler with the given task.
// If the call returns without panic, it simply returns the value, // If the call returns without panic, it simply returns the value,
// otherwise, it recovers from panic and returns an error. // otherwise, it recovers from panic and returns an error.
func perform(ctx context.Context, task *Task, h Handler) (err error) { func (p *processor) perform(ctx context.Context, task *Task) (err error) {
defer func() { defer func() {
if x := recover(); x != nil { if x := recover(); x != nil {
p.logger.Errorf("recovering from panic. See the stack trace below for details:\n%s", string(debug.Stack()))
err = fmt.Errorf("panic: %v", x) err = fmt.Errorf("panic: %v", x)
} }
}() }()
return h.ProcessTask(ctx, task) return p.handler.ProcessTask(ctx, task)
} }
// uniq dedupes elements and returns a slice of unique names of length l. // uniq dedupes elements and returns a slice of unique names of length l.

11
processor_test.go
View File

@ -564,7 +564,7 @@ func TestProcessorWithStrictPriority(t *testing.T) {
} }
} }
func TestPerform(t *testing.T) { func TestProcessorPerform(t *testing.T) {
tests := []struct { tests := []struct {
desc string desc string
handler HandlerFunc handler HandlerFunc
@ -596,9 +596,16 @@ func TestPerform(t *testing.T) {
wantErr: true, wantErr: true,
}, },
} }
// Note: We don't need to fully initialize the processor since we are only testing
// perform method.
p := newProcessor(processorParams{
logger: testLogger,
queues: defaultQueueConfig,
})
for _, tc := range tests { for _, tc := range tests {
got := perform(context.Background(), tc.task, tc.handler) p.handler = tc.handler
got := p.perform(context.Background(), tc.task)
if !tc.wantErr && got != nil { if !tc.wantErr && got != nil {
t.Errorf("%s: perform() = %v, want nil", tc.desc, got) t.Errorf("%s: perform() = %v, want nil", tc.desc, got)
continue continue