go.temporal.io/server/service/matching/task.go
394 LOC · 159 covered · 235 uncovered · 48 ranges · 863 concepts · 40 introducers · 414 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package matching
2
3
import (
4
"context"
5
"sync/atomic"
6
"time"
7
8
commonpb "go.temporal.io/api/common/v1"
9
taskqueuepb "go.temporal.io/api/taskqueue/v1"
10
deploymentspb "go.temporal.io/server/api/deployment/v1"
11
enumsspb "go.temporal.io/server/api/enums/v1"
12
"go.temporal.io/server/api/matchingservice/v1"
13
persistencespb "go.temporal.io/server/api/persistence/v1"
14
taskqueuespb "go.temporal.io/server/api/taskqueue/v1"
15
"go.temporal.io/server/common/namespace"
16
"google.golang.org/protobuf/types/known/timestamppb"
17
)
18
19
type (
20
// genericTaskInfo contains the info for an activity or workflow task
21
genericTaskInfo struct {
22
*persistencespb.AllocatedTaskInfo
23
completionFunc func(*internalTask, taskResponse)
24
}
25
// queryTaskInfo contains the info for a query task
26
queryTaskInfo struct {
27
taskID string
28
createTime *timestamppb.Timestamp
29
request *matchingservice.QueryWorkflowRequest
30
}
31
// nexusTaskInfo contains the info for a nexus task
32
nexusTaskInfo struct {
33
taskID string
34
createTime *timestamppb.Timestamp
35
deadline time.Time
36
operationDeadline time.Time
37
request *matchingservice.DispatchNexusTaskRequest
38
}
39
// startedTaskInfo contains info for any task received from
40
// another matching host. This type of task is already marked as started
41
startedTaskInfo struct {
42
workflowTaskInfo *matchingservice.PollWorkflowTaskQueueResponse
43
activityTaskInfo *matchingservice.PollActivityTaskQueueResponse
44
nexusTaskInfo *matchingservice.PollNexusTaskQueueResponse
45
}
46
// internalTask represents an activity, workflow, query or started (received from another host).
47
// this struct is more like a union and only one of [ query, event, forwarded ] is
48
// non-nil for any given task
49
// TODO(pri): after deprecating classic matcher, we can consolidate backlogCountHint, recycleToken,
50
// and removeFromMatcher into a single *physicalTaskQueueManager field.
51
internalTask struct {
52
event *genericTaskInfo // non-nil for activity or workflow task that's locally generated
53
query *queryTaskInfo // non-nil for a query task that's locally sync matched
54
nexus *nexusTaskInfo // non-nil for a nexus task that's locally sync matched
55
started *startedTaskInfo // non-nil for a task received from a parent partition which is already started
56
namespace namespace.Name
57
source enumsspb.TaskSource
58
responseC chan taskResponse // non-nil only where there is a caller waiting for response (sync match)
59
backlogCountHint func() int64
60
// forwardInfo contains information about forward source partition and versioning decisions made by it
61
// a parent partition receiving forwarded tasks makes no versioning decisions and only follows what the source
62
// partition instructed.
63
forwardInfo *taskqueuespb.TaskForwardInfo
64
// redirectInfo is only set when redirect rule is applied on the task. for forwarded tasks,
65
// this is populated based on forwardInfo (V2 versioning).
66
redirectInfo *taskqueuespb.BuildIdRedirectInfo
67
// redirectedFromBacklog is true if this task was redirectedFromBacklog from the backlog it was read from
68
// (V2 and V3 versioning).
69
redirectedFromBacklog bool
70
// pollerScalingDecision is assigned when the queue has advice to give to the poller about whether
71
// it should adjust its poller count
72
pollerScalingDecision *taskqueuepb.PollerScalingDecision
73
recycleToken func(*internalTask)
74
removeFromMatcher atomic.Pointer[func()]
75
// taskDispatchRevisionNumber represents the revision number used by the task and is
76
// max(taskDirectiveRevisionNumber, routingConfigRevisionNumber) for the task.
77
taskDispatchRevisionNumber int64
78
targetWorkerDeploymentVersion *deploymentspb.WorkerDeploymentVersion
79
80
// The following fields are for use by priMatcher/matcherData:
81
waitableMatchResult
82
forwardCtx context.Context // non-nil for sync match task only
83
// effectivePriority is initialized from an explicit task priority if present, or the
84
// default for the task queue. It can also be the special pollForwarderPriority (higher
85
// than normal priorities) to indicate the poll forwarder. In some other cases (e.g.
86
// migration) it may be adjusted from the explicit task priority.
87
// The scale of effectivePriority is 10× the normal scale to allow inserting forwards
88
// in between priority levels.
89
effectivePriority priorityKey
90
pollForwarderType pollForwarderType
91
}
92
93
// taskResponse is used to report the result of either a match with a local poller,
94
// or forwarding a task, query, or nexus task.
95
taskResponse struct {
96
// If forwarded is true, then forwardRes and forwardErr have the result of forwarding.
97
// If it's false, then startErr has the result of RecordTaskStarted.
98
forwarded bool
99
forwardRes any // note this may be a non-nil "any" containing a nil pointer
100
forwardErr error
101
startErr error
102
// dropReason, when set, marks a dropped backlog task;
103
// reader.completeTask records it in tasks_dropped.
104
dropReason dropReason
105
}
106
107
// taskFinishResult describes how a task finished. It is passed to internalTask.finish.
108
taskFinishResult struct {
109
// err is the result of RecordTaskStarted (or forwarding); nil on success or drop.
110
err error
111
// consumedToken reports whether the task consumed its rate-limit token (see finish).
112
consumedToken bool
113
// dropReason, when set, indicates the task is being dropped rather than dispatched
114
// and is recorded in tasks_dropped.
115
dropReason dropReason
116
}
117
)
118
119
var (
120
// sentinel values for task.removeFromMatcher
121
removeFuncNotAddedYet = func() {}
122
removeFuncEvicted = func() {}
123
)
124
126
>
if res.forwarded {
128
>
}
130
}
131
132
func newInternalTaskForSyncMatch(
133
info *persistencespb.TaskInfo,
134
forwardInfo *taskqueuespb.TaskForwardInfo,
135
taskDispatchRevisionNumber int64,
136
targetVersion *deploymentspb.WorkerDeploymentVersion,
138
>
var redirectInfo *taskqueuespb.BuildIdRedirectInfo
139
>
// if this task is not forwarded, source can only be history
140
>
source := enumsspb.TASK_SOURCE_HISTORY
141
>
if forwardInfo != nil {
142
>
// if task is forwarded, it may be history or backlog. setting based on forward info
message.pb.go ×1
143
>
source = forwardInfo.TaskSource
144
>
redirectInfo = forwardInfo.GetRedirectInfo()
145
>
}
147
>
event: &genericTaskInfo{
148
>
AllocatedTaskInfo: &persistencespb.AllocatedTaskInfo{
149
>
Data: info,
150
>
TaskId: syncMatchTaskId,
151
>
},
152
>
},
153
>
forwardInfo: forwardInfo,
154
>
source: source,
155
>
redirectInfo: redirectInfo,
156
>
responseC: make(chan taskResponse, 1),
157
>
158
>
taskDispatchRevisionNumber: taskDispatchRevisionNumber,
159
>
targetWorkerDeploymentVersion: targetVersion,
160
>
161
>
effectivePriority: effectivePriorityFactor * priorityKey(info.GetPriority().GetPriorityKey()),
162
>
}
163
}
164
165
func newInternalTaskFromBacklog(
166
info *persistencespb.AllocatedTaskInfo,
167
completionFunc func(*internalTask, taskResponse),
169
>
return &internalTask{
170
>
event: &genericTaskInfo{
171
>
AllocatedTaskInfo: info,
172
>
completionFunc: completionFunc,
173
>
},
174
>
source: enumsspb.TASK_SOURCE_DB_BACKLOG,
175
>
effectivePriority: effectivePriorityFactor * priorityKey(info.GetData().GetPriority().GetPriorityKey()),
176
>
}
177
>
}
178
179
func newInternalQueryTask(
180
taskID string,
181
request *matchingservice.QueryWorkflowRequest,
183
>
return &internalTask{
184
>
query: &queryTaskInfo{
185
>
taskID: taskID,
186
>
createTime: getCreateTime(request.GetForwardInfo()),
187
>
request: request,
188
>
},
189
>
forwardInfo: request.GetForwardInfo(),
190
>
responseC: make(chan taskResponse, 1),
191
>
source: enumsspb.TASK_SOURCE_HISTORY,
192
>
effectivePriority: effectivePriorityFactor * priorityKey(request.GetPriority().GetPriorityKey()),
193
>
}
194
>
}
195
197
>
if t := f.GetCreateTime(); t != nil {
198
return t
199
}
201
}
202
203
func newInternalNexusTask(
204
taskID string,
205
deadline time.Time,
206
operationDeadline time.Time,
207
request *matchingservice.DispatchNexusTaskRequest,
209
>
return &internalTask{
210
>
nexus: &nexusTaskInfo{
211
>
taskID: taskID,
212
>
createTime: getCreateTime(request.GetForwardInfo()),
213
>
deadline: deadline,
214
>
operationDeadline: operationDeadline,
215
>
request: request,
216
>
},
217
>
forwardInfo: request.GetForwardInfo(),
218
>
responseC: make(chan taskResponse, 1),
219
>
source: enumsspb.TASK_SOURCE_HISTORY,
220
>
}
221
>
}
222
224
>
return &internalTask{started: info}
225
>
}
226
228
>
return &internalTask{effectivePriority: p, pollForwarderType: t}
229
>
}
230
232
>
return task.pollForwarderType != notPollForwarder
233
>
}
234
235
// isQuery returns true if the underlying task is a query task
237
>
return task.query != nil
238
>
}
239
240
// isNexus returns true if the underlying task is a nexus task
242
>
return task.nexus != nil
243
>
}
244
245
// isStarted is true when this task is already marked as started
247
>
return task.started != nil
248
>
}
249
250
// isForwarded returns true if the underlying task is forwarded by a remote matching host
251
// forwarded tasks are already marked as started in history
253
>
return task.forwardInfo != nil
254
>
}
255
257
>
return task.responseC != nil
258
>
}
259
261
>
if task.forwardInfo.GetCreateTime() != nil {
269
>
}
270
271
return timestamppb.Now()
272
}
273
275
>
switch {
276
>
case task.event != nil:
277
>
return &commonpb.WorkflowExecution{WorkflowId: task.event.Data.GetWorkflowId(), RunId: task.event.Data.GetRunId()}
278
case task.query != nil:
279
return task.query.request.GetQueryRequest().GetExecution()
280
case task.started != nil && task.started.workflowTaskInfo != nil:
281
return task.started.workflowTaskInfo.WorkflowExecution
282
case task.started != nil && task.started.activityTaskInfo != nil:
283
return task.started.activityTaskInfo.WorkflowExecution
284
}
285
return &commonpb.WorkflowExecution{}
286
}
287
288
// pollWorkflowTaskQueueResponse returns the poll response for a workflow task that is
289
// already marked as started. This method should only be called when isStarted() is true
290
>
func (task *internalTask) pollWorkflowTaskQueueResponse() *matchingservice.PollWorkflowTaskQueueResponse {
task.go ×1
291
>
if task.isStarted() {
292
>
return task.started.workflowTaskInfo
293
>
}
294
return nil
295
}
296
297
// pollActivityTaskQueueResponse returns the poll response for an activity task that is
298
// already marked as started. This method should only be called when isStarted() is true
299
>
func (task *internalTask) pollActivityTaskQueueResponse() *matchingservice.PollActivityTaskQueueResponse {
task.go ×1
300
>
if task.isStarted() {
301
>
return task.started.activityTaskInfo
302
>
}
303
return nil
304
}
305
306
// pollNexusTaskQueueResponse returns the poll response for a nexus task that is ready for dispatching. This method
307
// should only be called when isStarted() is true
308
func (task *internalTask) pollNexusTaskQueueResponse() *matchingservice.PollNexusTaskQueueResponse {
309
if task.isStarted() {
310
if task.started.nexusTaskInfo.Response != nil {
311
task.started.nexusTaskInfo.Response.PollerScalingDecision = task.pollerScalingDecision
312
}
313
return task.started.nexusTaskInfo
314
}
315
return nil
316
}
317
318
// getResponse waits for a response on the task's response channel.
320
>
if task.responseC == nil {
321
return taskResponse{}, false
322
}
324
}
325
327
>
if task.event != nil {
331
>
}
332
// nexus tasks don't have priorities for now
334
}
335
337
>
return fairLevelFromAllocatedTask(task.event.AllocatedTaskInfo)
338
>
}
339
340
// resetMatcherState must be called before adding or re-adding a backlog task to priMatcher.
342
>
task.removeFromMatcher.Store(&removeFuncNotAddedYet)
343
>
}
344
345
// setRemoveFunc sets the function to remove the task from the matcher.
346
// It returns true if the task is still valid and the function was set,
347
// false if the task was evicted already and should not be added.
349
>
return task.removeFromMatcher.CompareAndSwap(&removeFuncNotAddedYet, &remove)
350
>
}
351
352
// setEvicted marks the task as evicted. If it was added to a matcher it will be removed.
354
>
remove := task.removeFromMatcher.Swap(&removeFuncEvicted)
355
>
(*remove)()
356
>
}
357
358
// finish marks a task as finished. Must be called after a poller picks up a task
359
// and marks it as started. If the task is unable to marked as started, then this
360
// method should be called with a non-nil error argument.
361
//
362
// If the task took a rate limit token and didn't "use" it by actually dispatching the task,
363
// finish will be called with consumedToken=false and task.recycleToken=clockedRateLimiter.RecycleToken,
364
// so finish will call the rate limiter's RecycleToken to give the unused token back to any process
365
// that is waiting on the token, if one exists.
366
//
367
// When a backlog task is being dropped rather than dispatched, set r.dropReason; it is
368
// carried on the taskResponse and counted in tasks_dropped by the backlog completion
369
// callback (reader.completeTask).
371
>
task.finishInternal(taskResponse{
372
>
startErr: r.err,
373
>
dropReason: r.dropReason,
374
>
}, r.consumedToken)
375
>
}
376
377
// finishForward must be called after forwarding a task.
378
>
func (task *internalTask) finishForward(forwardRes any, forwardErr error, consumedToken bool) {
task.go ×1
379
>
task.finishInternal(taskResponse{forwarded: true, forwardRes: forwardRes, forwardErr: forwardErr}, consumedToken)
380
>
}
381
383
>
if !consumedToken && task.recycleToken != nil {
385
>
}
386
389
>
task.responseC <- res
391
>
// TODO: this probably should not be done synchronously in PollWorkflow/ActivityTaskQueue
392
>
task.event.completionFunc(task, res)
393
}
394
}