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

Fix queue rate limiting busy loop #3111

Merged
merged 4 commits into from
Jul 22, 2022
Merged

Fix queue rate limiting busy loop #3111

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a6c47e9
Improve queue rate limiting
yycptt Jul 17, 2022
9c4f6be
fix static check
yycptt Jul 18, 2022
32279cb
ratelimiter backoff
yycptt Jul 18, 2022
f686353
revert order
yycptt Jul 22, 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
3 changes: 2 additions & 1 deletion service/frontend/dcRedirectionHandler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1656,14 +1656,15 @@ func (handler *DCRedirectionHandlerImpl) UpdateWorkflow(
switch {
case targetDC == handler.currentClusterName:
resp, err = handler.frontendHandler.UpdateWorkflow(ctx, request)
return err
default:
remoteClient, err := handler.clientBean.GetRemoteFrontendClient(targetDC)
if err != nil {
return err
}
resp, err = remoteClient.UpdateWorkflow(ctx, request)
return err
}
return err
})

return resp, err
Expand Down
9 changes: 4 additions & 5 deletions service/history/queueProcessor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,11 +216,6 @@ eventLoop:
}

func (p *queueProcessorBase) processBatch() {

if !p.verifyReschedulerSize() {
return
}

ctx, cancel := context.WithTimeout(context.Background(), loadQueueTaskThrottleRetryDelay)
if err := p.rateLimiter.Wait(ctx); err != nil {
cancel()
Expand All @@ -229,6 +224,10 @@ func (p *queueProcessorBase) processBatch() {
}
cancel()

if !p.verifyReschedulerSize() {
return
}

p.lastPollTime = p.timeSource.Now()
tasks, more, err := p.ackMgr.readQueueTasks()

Expand Down
4 changes: 3 additions & 1 deletion service/history/timerQueueProcessor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -414,12 +414,14 @@ func newQueueProcessorRateLimiter(
) quotas.RateLimiter {
return quotas.NewMultiRateLimiter(
[]quotas.RateLimiter{
// consume from host rate limiter as it's usually the one throttles the traffic
// and avoid unnecessary consume and cancel on shard level rate limiter
hostRateLimiter,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will one rogue shard exhaust all host quota?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My understanding is if later ratelimiter (the shard level one) can't pass, previous reservations will be cancelled.
https://github.com/temporalio/temporal/blob/master/common/quotas/multi_rate_limiter_impl.go#L106

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please test to verify.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My local test shows it works fine. But theoretically one rogue shard can still affect others since other shard can try to get a host token before the rouge shard cancels it. So I reverted the change here to be safe.

The busy loop issue is still fixed.

quotas.NewDefaultOutgoingRateLimiter(
func() float64 {
return float64(shardMaxPollRPS())
},
),
hostRateLimiter,
},
)
}
8 changes: 4 additions & 4 deletions service/history/timerQueueProcessorBase.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -299,10 +299,6 @@ eventLoop:
}

func (t *timerQueueProcessorBase) readAndFanoutTimerTasks() (*time.Time, error) {
if !t.verifyReschedulerSize() {
return nil, nil
}

ctx, cancel := context.WithTimeout(context.Background(), loadTimerTaskThrottleRetryDelay)
if err := t.rateLimiter.Wait(ctx); err != nil {
cancel()
Expand All @@ -311,6 +307,10 @@ func (t *timerQueueProcessorBase) readAndFanoutTimerTasks() (*time.Time, error)
}
cancel()

if !t.verifyReschedulerSize() {
return nil, nil
}

t.lastPollTime = t.timeSource.Now()
timerTasks, nextFireTime, moreTasks, err := t.timerQueueAckMgr.readTimerTasks()
if err != nil {
Expand Down