taskmatcher rewrite wip

temporalio · Jan 31, 2025 · fe8d0d7 · fe8d0d7
1 parent f195d5d
commit fe8d0d7
Show file tree

Hide file tree

Showing 17 changed files with 947 additions and 721 deletions.
diff --git a/common/dynamicconfig/constants.go b/common/dynamicconfig/constants.go
@@ -1070,11 +1070,6 @@ See DynamicRateLimitingParams comments for more details.`,
 		time.Minute,
 		`MatchingLongPollExpirationInterval is the long poll expiration interval in the matching service`,
 	)
-	MatchingSyncMatchWaitDuration = NewTaskQueueDurationSetting(
-		"matching.syncMatchWaitDuration",
-		200*time.Millisecond,
-		`MatchingSyncMatchWaitDuration is to wait time for sync match`,
-	)
 	MatchingHistoryMaxPageSize = NewNamespaceIntSetting(
 		"matching.historyMaxPageSize",
 		primitives.GetHistoryMaxPageSize,
@@ -1175,7 +1170,7 @@ for VERSIONED queues.`,
 		1,
 		`MatchingForwarderMaxOutstandingTasks is the max number of inflight addTask/queryTask from the forwarder`,
 	)
-	MatchingForwarderMaxRatePerSecond = NewTaskQueueIntSetting(
+	MatchingForwarderMaxRatePerSecond = NewTaskQueueFloatSetting(
 		"matching.forwarderMaxRatePerSecond",
 		10,
 		`MatchingForwarderMaxRatePerSecond is the max rate at which add/query can be forwarded`,

diff --git a/service/matching/ack_manager.go b/service/matching/ack_manager.go
@@ -26,7 +26,6 @@ package matching
 
 import (
 	"sync"
-	"sync/atomic"
 
 	"github.com/emirpasic/gods/maps/treemap"
 	godsutils "github.com/emirpasic/gods/utils"
@@ -41,7 +40,6 @@ type ackManager struct {
 	outstandingTasks *treemap.Map // TaskID->acked
 	readLevel        int64        // Maximum TaskID inserted into outstandingTasks
 	ackLevel         int64        // Maximum TaskID below which all tasks are acked
-	backlogCountHint atomic.Int64
 	logger           log.Logger
 }
 
@@ -68,7 +66,6 @@ func (m *ackManager) addTask(taskID int64) {
 		m.logger.Fatal("Already present in outstanding tasks", tag.TaskID(taskID))
 	}
 	m.outstandingTasks.Put(taskID, false)
-	m.backlogCountHint.Add(1)
 }
 
 func (m *ackManager) getReadLevel() int64 {
@@ -135,7 +132,6 @@ func (m *ackManager) completeTask(taskID int64) int64 {
 	// TODO the ack level management should be done by a dedicated coroutine
 	//  this is only a temporarily solution
 	m.outstandingTasks.Put(taskID, true)
-	m.backlogCountHint.Add(-1)
 
 	// Adjust the ack level as far as we can
 	var numberOfAckedTasks int64
@@ -153,7 +149,3 @@ func (m *ackManager) completeTask(taskID int64) int64 {
 	}
 	return m.ackLevel
 }
-
-func (m *ackManager) getBacklogCountHint() int64 {
-	return m.backlogCountHint.Load()
-}
diff --git a/service/matching/backlog_age_tracker.go b/service/matching/backlog_age_tracker.go
@@ -0,0 +1,49 @@
+package matching
+
+import (
+	"time"
+
+	"github.com/emirpasic/gods/maps/treemap"
+	godsutils "github.com/emirpasic/gods/utils"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+const emptyBacklogAge time.Duration = -1
+
+// backlogAgeTracker is not safe for concurrent use
+type backlogAgeTracker struct {
+	tree treemap.Map // unix nano as int64 -> int (count)
+}
+
+func newBacklogAgeTracker() backlogAgeTracker {
+	return backlogAgeTracker{tree: *treemap.NewWith(godsutils.Int64Comparator)}
+}
+
+// record adds or removes a task from the tracker.
+func (b backlogAgeTracker) record(ts *timestamppb.Timestamp, delta int) {
+	if ts == nil {
+		return
+	}
+
+	createTime := ts.AsTime().UnixNano()
+	count := delta
+	if prev, ok := b.tree.Get(createTime); ok {
+		count += prev.(int)
+	}
+	if count = max(0, count); count == 0 {
+		b.tree.Remove(createTime)
+	} else {
+		b.tree.Put(createTime, count)
+	}
+}
+
+// getBacklogAge returns the largest age in this backlog (age of oldest task),
+// or emptyBacklogAge if empty.
+func (b backlogAgeTracker) getAge() time.Duration {
+	if b.tree.Empty() {
+		return emptyBacklogAge
+	}
+	k, _ := b.tree.Min()
+	oldest := k.(int64)
+	return time.Since(time.Unix(0, oldest))
+}
diff --git a/service/matching/backlog_manager.go b/service/matching/backlog_manager.go
@@ -181,15 +181,12 @@ func (c *backlogManagerImpl) SpoolTask(taskInfo *persistencespb.TaskInfo) error
 	return err
 }
 
-func (c *backlogManagerImpl) processSpooledTask(
-	ctx context.Context,
-	task *internalTask,
-) error {
-	return c.pqMgr.ProcessSpooledTask(ctx, task)
+func (c *backlogManagerImpl) addSpooledTask(ctx context.Context, task *internalTask) error {
+	return c.pqMgr.AddSpooledTask(ctx, task)
 }
 
 func (c *backlogManagerImpl) BacklogCountHint() int64 {
-	return c.taskAckManager.getBacklogCountHint()
+	return c.taskReader.getLoadedTasks()
 }
 
 func (c *backlogManagerImpl) BacklogStatus() *taskqueuepb.TaskQueueStatus {

diff --git a/service/matching/config.go b/service/matching/config.go
@@ -32,6 +32,7 @@ import (
 
 	"go.temporal.io/server/common/backoff"
 	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/headers"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/common/tqid"
 )
@@ -46,7 +47,6 @@ type (
 		PersistencePerShardNamespaceMaxQPS   dynamicconfig.IntPropertyFnWithNamespaceFilter
 		PersistenceDynamicRateLimitingParams dynamicconfig.TypedPropertyFn[dynamicconfig.DynamicRateLimitingParams]
 		PersistenceQPSBurstRatio             dynamicconfig.FloatPropertyFn
-		SyncMatchWaitDuration                dynamicconfig.DurationPropertyFnWithTaskQueueFilter
 		RPS                                  dynamicconfig.IntPropertyFn
 		OperatorRPSRatio                     dynamicconfig.FloatPropertyFn
 		AlignMembershipChange                dynamicconfig.DurationPropertyFn
@@ -70,7 +70,7 @@ type (
 		BreakdownMetricsByBuildID                dynamicconfig.BoolPropertyFnWithTaskQueueFilter
 		ForwarderMaxOutstandingPolls             dynamicconfig.IntPropertyFnWithTaskQueueFilter
 		ForwarderMaxOutstandingTasks             dynamicconfig.IntPropertyFnWithTaskQueueFilter
-		ForwarderMaxRatePerSecond                dynamicconfig.IntPropertyFnWithTaskQueueFilter
+		ForwarderMaxRatePerSecond                dynamicconfig.FloatPropertyFnWithTaskQueueFilter
 		ForwarderMaxChildrenPerNode              dynamicconfig.IntPropertyFnWithTaskQueueFilter
 		VersionCompatibleSetLimitPerQueue        dynamicconfig.IntPropertyFnWithNamespaceFilter
 		VersionBuildIdLimitPerQueue              dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -124,13 +124,13 @@ type (
 	forwarderConfig struct {
 		ForwarderMaxOutstandingPolls func() int
 		ForwarderMaxOutstandingTasks func() int
-		ForwarderMaxRatePerSecond    func() int
+		ForwarderMaxRatePerSecond    func() float64
 		ForwarderMaxChildrenPerNode  func() int
 	}
 
 	taskQueueConfig struct {
 		forwarderConfig
-		SyncMatchWaitDuration        func() time.Duration
+		CallerInfo                   headers.CallerInfo
 		BacklogNegligibleAge         func() time.Duration
 		MaxWaitForPollerBeforeFwd    func() time.Duration
 		QueryPollerUnavailableWindow func() time.Duration
@@ -213,7 +213,6 @@ func NewConfig(
 		PersistencePerShardNamespaceMaxQPS:       dynamicconfig.DefaultPerShardNamespaceRPSMax,
 		PersistenceDynamicRateLimitingParams:     dynamicconfig.MatchingPersistenceDynamicRateLimitingParams.Get(dc),
 		PersistenceQPSBurstRatio:                 dynamicconfig.PersistenceQPSBurstRatio.Get(dc),
-		SyncMatchWaitDuration:                    dynamicconfig.MatchingSyncMatchWaitDuration.Get(dc),
 		LoadUserData:                             dynamicconfig.MatchingLoadUserData.Get(dc),
 		HistoryMaxPageSize:                       dynamicconfig.MatchingHistoryMaxPageSize.Get(dc),
 		EnableDeployments:                        dynamicconfig.EnableDeployments.Get(dc),
@@ -285,7 +284,8 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 	taskType := tq.TaskType()
 
 	return &taskQueueConfig{
-		RangeSize: config.RangeSize,
+		CallerInfo: headers.NewBackgroundCallerInfo(ns.String()),
+		RangeSize:  config.RangeSize,
 		GetTasksBatchSize: func() int {
 			return config.GetTasksBatchSize(ns.String(), taskQueueName, taskType)
 		},
@@ -298,9 +298,6 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 		MinTaskThrottlingBurstSize: func() int {
 			return config.MinTaskThrottlingBurstSize(ns.String(), taskQueueName, taskType)
 		},
-		SyncMatchWaitDuration: func() time.Duration {
-			return config.SyncMatchWaitDuration(ns.String(), taskQueueName, taskType)
-		},
 		BacklogNegligibleAge: func() time.Duration {
 			return config.BacklogNegligibleAge(ns.String(), taskQueueName, taskType)
 		},
@@ -353,7 +350,7 @@ func newTaskQueueConfig(tq *tqid.TaskQueue, config *Config, ns namespace.Name) *
 			ForwarderMaxOutstandingTasks: func() int {
 				return config.ForwarderMaxOutstandingTasks(ns.String(), taskQueueName, taskType)
 			},
-			ForwarderMaxRatePerSecond: func() int {
+			ForwarderMaxRatePerSecond: func() float64 {
 				return config.ForwarderMaxRatePerSecond(ns.String(), taskQueueName, taskType)
 			},
 			ForwarderMaxChildrenPerNode: func() int {