Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Use DC subscription in fifo scheduler (#6316)

Browse Source 
## What changed?
<!-- Describe what has changed in this PR -->
- Use DC subscriptions in fifo scheduler
- I will do sequential scheduler in a separate PR.

## Why?
<!-- Tell your future self why have you made these changes -->
- Simplify logic, avoid additional goroutines for polling dc update

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->

## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->

## Is hotfix candidate?
<!-- Is this PR a hotfix candidate or does it require a notification to
be sent to the broader community? (Yes/No) -->
This commit is contained in:
Yichao Yang
2024-07-22 13:37:54 -07:00
committed by GitHub
parent 514effa470
commit a509e5d903
5 changed files with 56 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
common/dynamicconfig/collection.go
View File

@@ -169,12 +169,12 @@ func (c *Collection) pollForChanges(ctx context.Context) error {
interval := DynamicConfigSubscriptionPollInterval.Get(c)
for ctx.Err() == nil {
util.InterruptibleSleep(ctx, interval())
c.pollOnce(ctx)
c.pollOnce()
}
return ctx.Err()
}
func (c *Collection) pollOnce(ctx context.Context) {
func (c *Collection) pollOnce() {
c.subscriptionLock.Lock()
defer c.subscriptionLock.Unlock()
if c.callbackPool == nil {

82
common/tasks/fifo_scheduler.go
View File

@@ -47,7 +47,7 @@ type (
// FIFOSchedulerOptions is the configs for FIFOScheduler
FIFOSchedulerOptions struct {
QueueSize int
WorkerCount dynamicconfig.IntPropertyFn
WorkerCount dynamicconfig.TypedSubscribable[int]
}
FIFOScheduler[T Task] struct {
@@ -56,10 +56,13 @@ type (
logger log.Logger
tasksChan chan T
shutdownChan chan struct{}
shutdownWG sync.WaitGroup
tasksChan chan T
shutdownWG sync.WaitGroup
workerLock sync.Mutex
workerShutdownCh []chan struct{}
workerCountSubscriptionCancelFn func()
}
)
@@ -74,8 +77,7 @@ func NewFIFOScheduler[T Task](
logger: logger,
tasksChan: make(chan T, options.QueueSize),
shutdownChan: make(chan struct{}),
tasksChan: make(chan T, options.QueueSize),
}
}
@@ -88,10 +90,9 @@ func (f *FIFOScheduler[T]) Start() {
return
}
f.startWorkers(f.options.WorkerCount())
f.shutdownWG.Add(1)
go f.workerMonitor()
initialWorkerCount, workerCountSubscriptionCancelFn := f.options.WorkerCount(f.updateWorkerCount)
f.workerCountSubscriptionCancelFn = workerCountSubscriptionCancelFn
f.updateWorkerCount(initialWorkerCount)
f.logger.Info("fifo scheduler started")
}
@@ -105,8 +106,8 @@ func (f *FIFOScheduler[T]) Stop() {
return
}
close(f.shutdownChan)
// must be called after the close of the shutdownChan
f.workerCountSubscriptionCancelFn()
f.updateWorkerCount(0)
f.drainTasks()
go func() {
@@ -136,39 +137,34 @@ func (f *FIFOScheduler[T]) TrySubmit(task T) bool {
}
}
func (f *FIFOScheduler[T]) workerMonitor() {
defer f.shutdownWG.Done()
func (f *FIFOScheduler[T]) updateWorkerCount(targetWorkerNum int) {
f.workerLock.Lock()
defer f.workerLock.Unlock()
timer := time.NewTimer(backoff.Jitter(defaultMonitorTickerDuration, defaultMonitorTickerJitter))
defer timer.Stop()
for {
select {
case <-f.shutdownChan:
f.stopWorkers(len(f.workerShutdownCh))
return
case <-timer.C:
timer.Reset(backoff.Jitter(defaultMonitorTickerDuration, defaultMonitorTickerJitter))
targetWorkerNum := f.options.WorkerCount()
if targetWorkerNum < 0 {
f.logger.Error("Target worker pool size is negative. Please fix the dynamic config.", tag.Key("worker-pool-size"), tag.Value(targetWorkerNum))
continue
}
currentWorkerNum := len(f.workerShutdownCh)
if targetWorkerNum == currentWorkerNum {
continue
}
if targetWorkerNum > currentWorkerNum {
f.startWorkers(targetWorkerNum - currentWorkerNum)
} else {
f.stopWorkers(currentWorkerNum - targetWorkerNum)
}
f.logger.Info("Update worker pool size", tag.Key("worker-pool-size"), tag.Value(targetWorkerNum))
}
if f.isStopped() {
// Always set the value to 0 when scheduler is stopped,
// in case there's a race condition between subscription callback invocation
// and the invocation made from Stop()
targetWorkerNum = 0
}
if targetWorkerNum < 0 {
f.logger.Error("Target worker pool size is negative. Please fix the dynamic config.", tag.Key("worker-pool-size"), tag.Value(targetWorkerNum))
return
}
currentWorkerNum := len(f.workerShutdownCh)
if targetWorkerNum == currentWorkerNum {
return
}
if targetWorkerNum > currentWorkerNum {
f.startWorkers(targetWorkerNum - currentWorkerNum)
} else {
f.stopWorkers(currentWorkerNum - targetWorkerNum)
}
f.logger.Info("Update worker pool size", tag.Key("worker-pool-size"), tag.Value(targetWorkerNum))
}
func (f *FIFOScheduler[T]) startWorkers(

7
common/tasks/fifo_scheduler_test.go
View File

@@ -35,7 +35,6 @@ import (
"github.com/stretchr/testify/suite"
"go.temporal.io/server/common/backoff"
"go.temporal.io/server/common/dynamicconfig"
"go.temporal.io/server/common/log"
)
@@ -219,8 +218,10 @@ func (s *fifoSchedulerSuite) TestStartStopWorkers() {
func (s *fifoSchedulerSuite) newTestProcessor() *FIFOScheduler[*MockTask] {
return NewFIFOScheduler[*MockTask](
&FIFOSchedulerOptions{
QueueSize: 1,
WorkerCount: dynamicconfig.GetIntPropertyFn(1),
QueueSize: 1,
WorkerCount: func(_ func(int)) (v int, cancel func()) {
return 1, func() {}
},
},
log.NewNoopLogger(),
)

20
service/history/configs/config.go
View File

@@ -127,7 +127,7 @@ type Config struct {
// TimerQueueProcessor settings
TimerTaskBatchSize dynamicconfig.IntPropertyFn
TimerProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
TimerProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
TimerProcessorSchedulerActiveRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
TimerProcessorSchedulerStandbyRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
TimerProcessorUpdateAckInterval dynamicconfig.DurationPropertyFn
@@ -141,11 +141,11 @@ type Config struct {
TimerQueueMaxReaderCount dynamicconfig.IntPropertyFn
RetentionTimerJitterDuration dynamicconfig.DurationPropertyFn
MemoryTimerProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
MemoryTimerProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
// TransferQueueProcessor settings
TransferTaskBatchSize dynamicconfig.IntPropertyFn
TransferProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
TransferProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
TransferProcessorSchedulerActiveRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
TransferProcessorSchedulerStandbyRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
TransferProcessorMaxPollRPS dynamicconfig.IntPropertyFn
@@ -303,7 +303,7 @@ type Config struct {
// ===== Visibility related =====
// VisibilityQueueProcessor settings
VisibilityTaskBatchSize dynamicconfig.IntPropertyFn
VisibilityProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
VisibilityProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
VisibilityProcessorSchedulerActiveRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
VisibilityProcessorSchedulerStandbyRoundRobinWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
VisibilityProcessorMaxPollRPS dynamicconfig.IntPropertyFn
@@ -333,7 +333,7 @@ type Config struct {
NamespaceCacheRefreshInterval dynamicconfig.DurationPropertyFn
// ArchivalQueueProcessor settings
ArchivalProcessorSchedulerWorkerCount dynamicconfig.IntPropertyFn
ArchivalProcessorSchedulerWorkerCount dynamicconfig.TypedSubscribable[int]
ArchivalProcessorMaxPollHostRPS dynamicconfig.IntPropertyFn
ArchivalTaskBatchSize dynamicconfig.IntPropertyFn
ArchivalProcessorPollBackoffInterval dynamicconfig.DurationPropertyFn
@@ -449,7 +449,7 @@ func NewConfig(
TaskSchedulerGlobalNamespaceMaxQPS: dynamicconfig.TaskSchedulerGlobalNamespaceMaxQPS.Get(dc),
TimerTaskBatchSize: dynamicconfig.TimerTaskBatchSize.Get(dc),
TimerProcessorSchedulerWorkerCount: dynamicconfig.TimerProcessorSchedulerWorkerCount.Get(dc),
TimerProcessorSchedulerWorkerCount: dynamicconfig.TimerProcessorSchedulerWorkerCount.Subscribe(dc),
TimerProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.TimerProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
TimerProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.TimerProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
TimerProcessorUpdateAckInterval: dynamicconfig.TimerProcessorUpdateAckInterval.Get(dc),
@@ -463,10 +463,10 @@ func NewConfig(
TransferQueueMaxReaderCount: dynamicconfig.TransferQueueMaxReaderCount.Get(dc),
RetentionTimerJitterDuration: dynamicconfig.RetentionTimerJitterDuration.Get(dc),
MemoryTimerProcessorSchedulerWorkerCount: dynamicconfig.MemoryTimerProcessorSchedulerWorkerCount.Get(dc),
MemoryTimerProcessorSchedulerWorkerCount: dynamicconfig.MemoryTimerProcessorSchedulerWorkerCount.Subscribe(dc),
TransferTaskBatchSize: dynamicconfig.TransferTaskBatchSize.Get(dc),
TransferProcessorSchedulerWorkerCount: dynamicconfig.TransferProcessorSchedulerWorkerCount.Get(dc),
TransferProcessorSchedulerWorkerCount: dynamicconfig.TransferProcessorSchedulerWorkerCount.Subscribe(dc),
TransferProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.TransferProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
TransferProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.TransferProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
TransferProcessorMaxPollRPS: dynamicconfig.TransferProcessorMaxPollRPS.Get(dc),
@@ -593,7 +593,7 @@ func NewConfig(
VisibilityTaskBatchSize: dynamicconfig.VisibilityTaskBatchSize.Get(dc),
VisibilityProcessorMaxPollRPS: dynamicconfig.VisibilityProcessorMaxPollRPS.Get(dc),
VisibilityProcessorMaxPollHostRPS: dynamicconfig.VisibilityProcessorMaxPollHostRPS.Get(dc),
VisibilityProcessorSchedulerWorkerCount: dynamicconfig.VisibilityProcessorSchedulerWorkerCount.Get(dc),
VisibilityProcessorSchedulerWorkerCount: dynamicconfig.VisibilityProcessorSchedulerWorkerCount.Subscribe(dc),
VisibilityProcessorSchedulerActiveRoundRobinWeights: dynamicconfig.VisibilityProcessorSchedulerActiveRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultActiveTaskPriorityWeight)).Get(dc),
VisibilityProcessorSchedulerStandbyRoundRobinWeights: dynamicconfig.VisibilityProcessorSchedulerStandbyRoundRobinWeights.WithDefault(ConvertWeightsToDynamicConfigValue(DefaultStandbyTaskPriorityWeight)).Get(dc),
VisibilityProcessorMaxPollInterval: dynamicconfig.VisibilityProcessorMaxPollInterval.Get(dc),
@@ -628,7 +628,7 @@ func NewConfig(
ArchivalTaskBatchSize: dynamicconfig.ArchivalTaskBatchSize.Get(dc),
ArchivalProcessorMaxPollRPS: dynamicconfig.ArchivalProcessorMaxPollRPS.Get(dc),
ArchivalProcessorMaxPollHostRPS: dynamicconfig.ArchivalProcessorMaxPollHostRPS.Get(dc),
ArchivalProcessorSchedulerWorkerCount: dynamicconfig.ArchivalProcessorSchedulerWorkerCount.Get(dc),
ArchivalProcessorSchedulerWorkerCount: dynamicconfig.ArchivalProcessorSchedulerWorkerCount.Subscribe(dc),
ArchivalProcessorMaxPollInterval: dynamicconfig.ArchivalProcessorMaxPollInterval.Get(dc),
ArchivalProcessorMaxPollIntervalJitterCoefficient: dynamicconfig.ArchivalProcessorMaxPollIntervalJitterCoefficient.Get(dc),
ArchivalProcessorUpdateAckInterval: dynamicconfig.ArchivalProcessorUpdateAckInterval.Get(dc),

2
service/history/queues/scheduler.go
View File

@@ -70,7 +70,7 @@ type (
ChannelWeightFn = tasks.ChannelWeightFn[TaskChannelKey]
SchedulerOptions struct {
WorkerCount dynamicconfig.IntPropertyFn
WorkerCount dynamicconfig.TypedSubscribable[int]
ActiveNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
StandbyNamespaceWeights dynamicconfig.MapPropertyFnWithNamespaceFilter
}