130
partitionMgr *taskQueuePartitionManagerImpl,
131
queue *PhysicalTaskQueueKey,
133
>
e := partitionMgr.engine
134
>
config := partitionMgr.config
135
>
versionTagValue := queue.Version().MetricsTagValue()
136
>
buildIDTag := tag.WorkerVersion(versionTagValue)
137
>
taggedMetricsHandler := partitionMgr.metricsHandler.WithTags(
138
>
metrics.OperationTag(metrics.MatchingTaskQueueMgrScope),
139
>
metrics.WorkerVersionTag(versionTagValue, config.BreakdownMetricsByBuildID()),
140
>
metrics.WorkerDeploymentNameTag(queue.Version().Deployment().GetSeriesName(), config.BreakdownMetricsByBuildID()),
141
>
metrics.WorkerDeploymentBuildIDTag(queue.Version().Deployment().GetBuildId(), config.BreakdownMetricsByBuildID()),
142
>
)
143
>
144
>
tqCtx, tqCancel := context.WithCancel(partitionMgr.callerInfoContext(context.Background()))
145
>
146
>
// We multiply by a big number so that we can later divide it by the number of pollers when grabbing permits,
147
>
// to allow us to make more decisions per second when there are more pollers.
148
>
pollerScalingRateLimitFn := func() float64 {
149
>
return config.PollerScalingDecisionsPerSecond() * 1e6
150
>
}
151
>
pqMgr := &physicalTaskQueueManagerImpl{
152
>
status: common.DaemonStatusInitialized,
153
>
partitionMgr: partitionMgr,
154
>
queue: queue,
155
>
config: config,
156
>
tqCtx: tqCtx,
157
>
tqCtxCancel: tqCancel,
158
>
namespaceRegistry: e.namespaceRegistry,
159
>
matchingClient: e.matchingRawClient,
160
>
clusterMeta: e.clusterMeta,
161
>
metricsHandler: taggedMetricsHandler,
162
>
tasksAdded: make(map[priorityKey]*taskTracker),
163
>
tasksDispatched: make(map[priorityKey]*taskTracker),
164
>
tasksRateLimited: e.newTaskTracker(),
165
>
pollerScalingRateLimiter: quotas.NewDefaultOutgoingRateLimiter(pollerScalingRateLimitFn),
166
>
deploymentRegistrationCh: make(chan struct{}, 1),
167
>
}
168
>
pqMgr.deploymentRegistrationCh <- struct{}{} // seed
169
>
170
>
pqMgr.pollerHistory = newPollerHistory(partitionMgr.config.PollerHistoryTTL())
171
>
172
>
pqMgr.liveness = newLiveness(
173
>
clock.NewRealTimeSource(),
174
>
config.MaxTaskQueueIdleTime,
175
>
func() { pqMgr.UnloadFromPartitionManager(unloadCauseIdle) },
176
)
177
179
>
tqCtx,
180
>
pqMgr.clusterMeta,
181
>
pqMgr.namespaceRegistry,
182
>
pqMgr.partitionMgr.engine.historyClient,
183
>
)
184
>
185
>
switch {
186
case config.EnableFairness:
187
pqMgr.logger = log.With(partitionMgr.logger, buildIDTag, backlogTagFairness)