Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

feat: Implement retryable job via timer #25

Merged
merged 5 commits into from
Nov 25, 2022
Merged

feat: Implement retryable job via timer #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6c0670a
feat: Implement retryable job via timer
DNK90 Nov 24, 2022
3e86307
Add ctx.Done handler to `PrepareRetryableJob` to update pending retry…
DNK90 Nov 24, 2022
8494d48
chore: Remove checking duplication on prepareJob
DNK90 Nov 25, 2022
e593361
chore: return if storing job to db failed when doing graceful shutdown
DNK90 Nov 25, 2022
a0de60c
fix: delete wrong job push
Nov 25, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 58 additions & 35 deletions pool.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ import (
"github.com/ethereum/go-ethereum/log"
"gorm.io/gorm"
"runtime/debug"
"sync"
"sync/atomic"
"time"
)
Expand All @@ -34,13 +33,12 @@ type Pool struct {

// JobChan receives new job
JobChan chan JobHandler
RetryJobChan chan JobHandler
SuccessJobChan chan JobHandler
FailedJobChan chan JobHandler
PrepareJobChan chan JobHandler

jobId int32
processedJobs sync.Map

jobId int32
MaxQueueSize int

store stores.MainStore
Expand Down Expand Up @@ -70,6 +68,7 @@ func NewPool(ctx context.Context, cfg *Config, db *gorm.DB, workers []Worker) *P
pool.PrepareJobChan = make(chan JobHandler, pool.MaxQueueSize)
pool.SuccessJobChan = make(chan JobHandler, pool.MaxQueueSize)
pool.FailedJobChan = make(chan JobHandler, pool.MaxQueueSize)
pool.RetryJobChan = make(chan JobHandler, pool.MaxQueueSize)
pool.Queue = make(chan chan JobHandler, pool.MaxQueueSize)

if workers != nil {
Expand Down Expand Up @@ -111,14 +110,20 @@ func (p *Pool) startWorker(w Worker) {
continue
}
log.Info("processing job", "id", job.GetID(), "nextTry", job.GetNextTry(), "retryCount", job.GetRetryCount(), "type", job.GetType())
if job.GetNextTry() == 0 || job.GetNextTry() <= time.Now().Unix() {
metrics.Pusher.IncrGauge(metrics.ProcessingJobMetric, 1)
w.ProcessJob(job)
metrics.Pusher.IncrGauge(metrics.ProcessingJobMetric, -1)
continue
metrics.Pusher.IncrGauge(metrics.ProcessingJobMetric, 1)
if err := w.ProcessJob(job); err != nil {
// update try and next retry time
if job.GetRetryCount()+1 > job.GetMaxTry() {
log.Info("[Pool][processJob] job reaches its maxTry", "jobTransaction", job.GetTransaction().GetHash().Hex())
p.FailedJobChan <- job
continue
}
job.IncreaseRetryCount()
job.UpdateNextTry(time.Now().Unix() + int64(job.GetRetryCount()*job.GetBackOff()))
// send to retry job chan
p.RetryJobChan <- job
}
// push the job back to mainChan
w.PoolChannel() <- job
metrics.Pusher.IncrGauge(metrics.ProcessingJobMetric, -1)
case <-w.Context().Done():
w.Close()
return
Expand All @@ -140,6 +145,8 @@ func (p *Pool) Start(closeFunc func()) {
p.processSuccessJob(job)
case job := <-p.FailedJobChan:
p.processFailedJob(job)
case job := <-p.RetryJobChan:
go p.PrepareRetryableJob(job)
case job := <-p.PrepareJobChan:
if job == nil {
continue
Expand All @@ -156,11 +163,6 @@ func (p *Pool) Start(closeFunc func()) {
if job == nil {
continue
}
// get 1 workerCh from queue and push job to this channel
hash := job.Hash()
if _, ok := p.processedJobs.LoadOrStore(hash, struct{}{}); ok {
continue
}
log.Info("[Pool] jobChan received a job", "jobId", job.GetID(), "nextTry", job.GetNextTry(), "type", job.GetType())
workerCh := <-p.Queue
workerCh <- job
Expand All @@ -185,6 +187,19 @@ func (p *Pool) Start(closeFunc func()) {
}
}

for {
log.Info("checking retrying jobs")
if len(p.RetryJobChan) == 0 {
break
}
job, more := <-p.RetryJobChan
if !more {
break
}
// update job
p.updateRetryingJob(job)
}

// update all success jobs
for {
log.Info("checking successJobChan")
Expand Down Expand Up @@ -216,8 +231,9 @@ func (p *Pool) Start(closeFunc func()) {
close(p.JobChan)
close(p.SuccessJobChan)
close(p.FailedJobChan)
close(p.RetryJobChan)
close(p.Queue)

log.Info("finish closing pool")
// send signal to stop the program
p.stop <- struct{}{}
Expand All @@ -226,32 +242,43 @@ func (p *Pool) Start(closeFunc func()) {
}
}

func (p *Pool) PrepareRetryableJob(job JobHandler) {
dur := time.Until(time.Unix(job.GetNextTry(), 0))
if dur <= 0 {
return
}
timer := time.NewTimer(dur)
select {
case <-timer.C:
p.PrepareJobChan <- job
case <-p.ctx.Done():
p.updateRetryingJob(job)
}
}

// PrepareJob saves new job to database
func (p *Pool) PrepareJob(job JobHandler) error {
if job == nil {
return nil
}
// deduplication: get hash from data and type and check if it exists in `processedJobs` or not.
hash := p.utils.RlpHash(struct {
Data []byte
Type int
}{
Data: job.GetData(),
Type: job.GetType(),
})
log.Info("[PrepareJobChan] preparing new job", "job", job.String(), "hash", hash.Hex())
if _, ok := p.processedJobs.Load(hash); ok {
return nil
}
// save job to db if id = 0
if job.GetID() == 0 {
return job.Save()
}
// cache above hash to `processedJobs`
p.processedJobs.Store(hash, struct{}{})
return nil
}

func (p *Pool) updateRetryingJob(job JobHandler) {
if job == nil {
return
}
log.Info("process retryable job", "id", job.GetID())
if err := job.Update(stores.STATUS_PENDING); err != nil {
log.Error("[Pool] failed on updating retrying job", "err", err, "jobType", job.GetType(), "tx", job.GetTransaction().GetHash().Hex())
return
}
}

// processSuccessJob updates job's status to `done` to database
func (p *Pool) processSuccessJob(job JobHandler) {
if job == nil {
Expand All @@ -261,8 +288,6 @@ func (p *Pool) processSuccessJob(job JobHandler) {
log.Info("process job success", "id", job.GetID())
if err := job.Update(stores.STATUS_DONE); err != nil {
log.Error("[Pool] failed on updating success job", "err", err, "jobType", job.GetType(), "tx", job.GetTransaction().GetHash().Hex())
// send back job to successJobChan
p.SuccessJobChan <- job
return
}
}
Expand All @@ -276,8 +301,6 @@ func (p *Pool) processFailedJob(job JobHandler) {
log.Info("process job failed", "id", job.GetID())
if err := job.Update(stores.STATUS_FAILED); err != nil {
log.Error("[Pool] failed on updating failed job", "err", err, "jobType", job.GetType(), "tx", job.GetTransaction().GetHash().Hex())
// send back job to failedJobChan
p.FailedJobChan <- job
return
}
}
Expand Down
30 changes: 6 additions & 24 deletions worker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,12 @@ import (
"github.com/axieinfinity/bridge-core/utils"
"github.com/ethereum/go-ethereum/log"
"sync/atomic"
"time"
)

type Worker interface {
Context() context.Context
Close()
ProcessJob(job JobHandler)
ProcessJob(job JobHandler) error
IsClose() bool
Channel() chan JobHandler
PoolChannel() chan<- JobHandler
Expand Down Expand Up @@ -63,36 +62,19 @@ func (w *BridgeWorker) String() string {
return fmt.Sprintf("{ id: %d, currentSize: %d }", w.id, len(w.workerChan))
}

func (w *BridgeWorker) ProcessJob(job JobHandler) {
var (
err error
val []byte
)

val, err = job.Process()
func (w *BridgeWorker) ProcessJob(job JobHandler) error {
val, err := job.Process()
if err != nil {
log.Error("[BridgeWorker] failed while processing job", "id", job.GetID(), "err", err)
goto ERROR
metrics.Pusher.IncrCounter(metrics.ProcessedFailedJobMetric, 1)
return err
}
if job.GetType() == ListenHandler && job.GetSubscriptionName() != "" {
job.GetListener().SendCallbackJobs(w.listeners, job.GetSubscriptionName(), job.GetTransaction(), val)
}
metrics.Pusher.IncrCounter(metrics.ProcessedSuccessJobMetric, 1)
w.successChan <- job
return
ERROR:
metrics.Pusher.IncrCounter(metrics.ProcessedFailedJobMetric, 1)

if job.GetRetryCount()+1 > job.GetMaxTry() {
log.Info("[BridgeWorker][processJob] job reaches its maxTry", "jobTransaction", job.GetTransaction().GetHash().Hex())
w.failedChan <- job
return
}
job.IncreaseRetryCount()
job.UpdateNextTry(time.Now().Unix() + int64(job.GetRetryCount()*job.GetBackOff()))
// push the job back to mainChan
w.mainChan <- job
log.Info("[BridgeWorker][processJob] job failed, added back to jobChan", "id", job.GetID(), "retryCount", job.GetRetryCount(), "nextTry", job.GetNextTry())
return nil
}

func (w *BridgeWorker) Context() context.Context {
Expand Down