Skip to content

auto enabling priority and fairness #8650

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

Draft
wants to merge 13 commits into
base: main
Choose a base branch
from
Draft

auto enabling priority and fairness #8650

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
be61432
add code for auto enabling priority and fairness
majiru Nov 14, 2025
87ae80e
use reasonable timeout for newTaskQueuePartitionManager in tests instead
majiru Nov 14, 2025
36864fa
fix linting issues
majiru Nov 15, 2025
b96901b
Refactor OnChange to be set after initialization through a seperate
majiru Nov 17, 2025
9a41e9a
Push taskQueuePartitionManager initialization into the background
majiru Nov 17, 2025
994cedb
Address comments made in review and simplify implementation
majiru Nov 18, 2025
dd19b25
Fix physicalTaskQueueManager
majiru Nov 18, 2025
0d9eb51
More lint fixes
majiru Nov 18, 2025
7a81262
Guard the defaultQueue behind the future
majiru Nov 19, 2025
f525e0f
We need to wait on ready before checking the cancellation of our
majiru Nov 19, 2025
1bdd110
make linter happy w.r.t deprecated functions
majiru Nov 19, 2025
7b38daf
Rework how we signal and handle waiting on initialization
majiru Nov 21, 2025
c18ce5f
Small linting issue
majiru Nov 21, 2025
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
204 changes: 139 additions & 65 deletions api/persistence/v1/task_queues.pb.go
View file
Open in desktop

Large diffs are not rendered by default.

5 changes: 5 additions & 0 deletions common/dynamicconfig/constants.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1382,6 +1382,11 @@ second per poller by one physical queue manager`,
The metric has 2 dimensions: namespace_id and plugin_name. Disabled by default as this is
an optional feature and also requires a metrics collection system that can handle higher cardinalities.`,
)
MatchingAutoEnableV2 = NewTaskQueueBoolSetting(
"matching.autoEnableV2",
false,
`MatchingAutoEnableV2 automatically enables fairness when a fairness or priority key is seen`,
)

// keys for history

Expand Down
7 changes: 7 additions & 0 deletions proto/internal/temporal/server/api/persistence/v1/task_queues.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,6 +134,13 @@ message TaskQueueTypeUserData {
DeploymentData deployment_data = 1;

temporal.api.taskqueue.v1.TaskQueueConfig config = 2;

enum FairnessState {
FAIRNESS_STATE_UNSPECIFIED = 0;
FAIRNESS_STATE_V1 = 1;
FAIRNESS_STATE_V2 = 2;
};
Comment on lines +138 to +142
Copy link
Contributor

@dnr dnr Nov 20, 2025

Choose a reason for hiding this comment

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

hmm, I think we usually put enums at the top-level. but I guess this is fine, we don't expect to use it anywhere else.

FairnessState fairness_state = 3;
}

// Container for all persistent user provided data for a task queue family.
Expand Down
6 changes: 6 additions & 0 deletions service/matching/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ type (
MembershipUnloadDelay dynamicconfig.DurationPropertyFn
TaskQueueInfoByBuildIdTTL dynamicconfig.DurationPropertyFnWithTaskQueueFilter
PriorityLevels dynamicconfig.IntPropertyFnWithTaskQueueFilter
AutoEnable dynamicconfig.BoolPropertyFnWithTaskQueueFilter

RateLimiterRefreshInterval time.Duration
FairnessKeyRateLimitCacheSize dynamicconfig.IntPropertyFnWithTaskQueueFilter
Expand Down Expand Up @@ -149,6 +150,7 @@ type (
EnableFairness bool
EnableFairnessSub func(func(bool)) (bool, func())
EnableMigration func() bool
AutoEnable func() bool
GetTasksBatchSize func() int
GetTasksReloadAt func() int
UpdateAckInterval func() time.Duration
Expand Down Expand Up @@ -308,6 +310,7 @@ func NewConfig(
FairnessKeyRateLimitCacheSize: dynamicconfig.MatchingFairnessKeyRateLimitCacheSize.Get(dc),
MaxFairnessKeyWeightOverrides: dynamicconfig.MatchingMaxFairnessKeyWeightOverrides.Get(dc),
MaxIDLengthLimit: dynamicconfig.MaxIDLengthLimit.Get(dc),
AutoEnable: dynamicconfig.MatchingAutoEnableV2.Get(dc),

AdminNamespaceToPartitionDispatchRate: dynamicconfig.AdminMatchingNamespaceToPartitionDispatchRate.Get(dc),
AdminNamespaceToPartitionRateSub: dynamicconfig.AdminMatchingNamespaceToPartitionDispatchRate.Subscribe(dc),
Expand Down Expand Up @@ -355,6 +358,9 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
EnableMigration: func() bool {
return config.EnableMigration(ns.String(), taskQueueName, taskType)
},
AutoEnable: func() bool {
return config.AutoEnable(ns.String(), taskQueueName, taskType)
},
GetTasksBatchSize: func() int {
return config.GetTasksBatchSize(ns.String(), taskQueueName, taskType)
},
Expand Down
2 changes: 1 addition & 1 deletion service/matching/matching_engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -430,7 +430,7 @@ func (e *matchingEngineImpl) getTaskQueuePartitionManager(
tqConfig.loadCause = loadCause
logger, throttledLogger, metricsHandler := e.loggerAndMetricsForPartition(namespaceEntry, partition, tqConfig)
onFatalErr := func(cause unloadCause) { newPM.unloadFromEngine(cause) }
onUserDataChanged := func() { newPM.userDataChanged() }
onUserDataChanged := func(to *persistencespb.VersionedTaskQueueUserData) { newPM.userDataChanged(to) }
userDataManager := newUserDataManager(
e.taskManager,
e.matchingRawClient,
Expand Down
15 changes: 8 additions & 7 deletions service/matching/matching_engine_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2119,8 +2119,7 @@ func (s *matchingEngineSuite) TestTaskQueueManagerGetTaskBatch() {
s.NoError(err)
}

tlMgr, ok := s.matchingEngine.partitions[dbq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
s.True(ok, "taskQueueManger doesn't implement taskQueuePartitionManager interface")
tlMgr := s.getPhysicalTaskQueueManagerImpl(dbq)
s.EqualValues(taskCount, s.taskManager.getTaskCount(dbq))

// wait until all tasks are read by the task pump and enqueued into the in-memory buffer
Expand Down Expand Up @@ -2254,8 +2253,7 @@ func (s *matchingEngineSuite) TestTaskExpiryAndCompletion() {
s.NoError(err)
}

tlMgr, ok := s.matchingEngine.partitions[dbq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
s.True(ok, "failed to load task queue")
tlMgr := s.getPhysicalTaskQueueManagerImpl(dbq)
s.EqualValues(taskCount, s.taskManager.getTaskCount(dbq))
blm := tlMgr.backlogMgr.(*backlogManagerImpl)

Expand Down Expand Up @@ -2988,7 +2986,9 @@ func (s *matchingEngineSuite) TestUpdatePhysicalTaskQueueGauge_UnVersioned() {
// the size of the map to 1 and it's counter to 1.
s.PhysicalQueueMetricValidator(capture, 1, 1)

tlmImpl, ok := tqm.(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
defaultQ, err := tqm.(*taskQueuePartitionManagerImpl).defaultQueueFuture.Get(context.Background())
s.Require().NoError(err)
tlmImpl := defaultQ.(*physicalTaskQueueManagerImpl)
s.True(ok)

s.matchingEngine.updatePhysicalTaskQueueGauge(s.ns, prtn, tlmImpl.queue.version, 1)
Expand Down Expand Up @@ -3223,8 +3223,9 @@ func (s *matchingEngineSuite) pollWorkflowTasksConcurrently(

// getPhysicalTaskQueueManagerImpl extracts the physicalTaskQueueManagerImpl for the given PhysicalTaskQueueKey
func (s *matchingEngineSuite) getPhysicalTaskQueueManagerImpl(ptq *PhysicalTaskQueueKey) *physicalTaskQueueManagerImpl {
pgMgr, ok := s.matchingEngine.partitions[ptq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueue.(*physicalTaskQueueManagerImpl)
s.True(ok, "taskQueueManger doesn't implement taskQueuePartitionManager interface")
defaultQ, err := s.matchingEngine.partitions[ptq.Partition().Key()].(*taskQueuePartitionManagerImpl).defaultQueueFuture.Get(context.Background())
s.Require().NoError(err)
pgMgr := defaultQ.(*physicalTaskQueueManagerImpl)
return pgMgr
}

Expand Down
15 changes: 13 additions & 2 deletions service/matching/physical_task_queue_manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,16 @@ func (s *PhysicalTaskQueueManagerTestSuite) SetupTest() {
s.NoError(err)
engine.partitions[prtn.Key()] = prtnMgr

if s.fairness {
prtnMgr.config.NewMatcher = true
prtnMgr.config.EnableFairness = true
} else if s.newMatcher {
prtnMgr.config.NewMatcher = true
}
Comment on lines +94 to +99
Copy link
Contributor

@dnr dnr Nov 20, 2025

Choose a reason for hiding this comment

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

hmm, the stuff on line 70-74 doesn't take care of this? if we do need this, maybe we don't need that?

reading this whole function again... it duplicates a lot of real code to set things up. I wish we could consolidate. but probably not now.


s.tqMgr, err = newPhysicalTaskQueueManager(prtnMgr, s.physicalTaskQueueKey)
s.NoError(err)
prtnMgr.defaultQueue = s.tqMgr
prtnMgr.defaultQueueFuture.Set(s.tqMgr, nil)
}

/*
Expand Down Expand Up @@ -334,13 +341,17 @@ func (s *PhysicalTaskQueueManagerTestSuite) TestAddTaskStandby() {
s.tqMgr.namespaceRegistry = mockNamespaceCache

s.tqMgr.Start()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond)
err := s.tqMgr.WaitUntilInitialized(ctx)
s.Require().NoError(err)
defer s.tqMgr.Stop(unloadCauseShuttingDown)
cancel()

// stop taskWriter so that we can check if there's any call to it
// otherwise the task persist process is async and hard to test
s.tqMgr.tqCtxCancel()

err := s.tqMgr.SpoolTask(&persistencespb.TaskInfo{
err = s.tqMgr.SpoolTask(&persistencespb.TaskInfo{
CreateTime: timestamp.TimePtr(time.Now().UTC()),
})
s.Equal(errShutdown, err) // task writer was stopped above
Expand Down
Loading
Loading