go.temporal.io/server/service/matching/matcher.go
655 LOC · 378 covered · 277 uncovered · 130 ranges · 300 concepts · 93 introducers · 202 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
"errors"
6
"math"
7
"sync"
8
"sync/atomic"
9
"time"
10
11
enumsspb "go.temporal.io/server/api/enums/v1"
12
"go.temporal.io/server/api/matchingservice/v1"
13
"go.temporal.io/server/common/metrics"
14
"go.temporal.io/server/common/primitives/timestamp"
15
"go.temporal.io/server/common/quotas"
16
"google.golang.org/grpc/codes"
17
"google.golang.org/grpc/status"
18
)
19
20
// TaskMatcher matches a task producer with a task consumer
21
// Producers are usually rpc calls from history or taskReader
22
// that drains backlog from db. Consumers are the task queue pollers
23
type TaskMatcher struct {
24
config *taskQueueConfig
25
26
// synchronous task channel to match producer/consumer
27
taskC chan *internalTask
28
// synchronous task channel to match query task - the reason to have a
29
// separate channel for this is that there are cases where consumers
30
// are interested in queryTasks but not others. One example is when a
31
// namespace is not active in a cluster.
32
queryTaskC chan *internalTask
33
// channel closed when task queue is closed, to interrupt pollers
34
closeC chan struct{}
35
36
// rateLimiter that limits the rate at which tasks can be dispatched to consumers
37
rateLimiter quotas.RateLimiter
38
39
fwdr *Forwarder
40
metricsHandler metrics.Handler // namespace metric scope
41
backlogTasksCreateTime map[int64]int // task creation time (unix nanos) -> number of tasks with that time
42
backlogTasksLock sync.Mutex
43
lastPoller atomic.Int64 // unix nanos of most recent poll start time
44
waitingPollerCount atomic.Int64
45
}
46
47
var (
48
// Sentinel error to redirect while blocked in matcher.
49
errInterrupted = errors.New("interrupted offer")
50
errNoRecentPoller = status.Error(codes.FailedPrecondition, "no poller seen for task queue recently, worker may be down")
51
)
52
53
// newTaskMatcher returns a task matcher instance. The returned instance can be used by task producers and consumers to
54
// find a match. Both sync matches and non-sync matches should use this implementation
55
>
func newTaskMatcher(config *taskQueueConfig, fwdr *Forwarder, metricsHandler metrics.Handler, rateLimiter quotas.RateLimiter) *TaskMatcher {
matcher.go ×1
56
>
return &TaskMatcher{
57
>
config: config,
58
>
rateLimiter: rateLimiter,
59
>
metricsHandler: metricsHandler,
60
>
fwdr: fwdr,
61
>
taskC: make(chan *internalTask),
62
>
queryTaskC: make(chan *internalTask),
63
>
closeC: make(chan struct{}),
64
>
backlogTasksCreateTime: make(map[int64]int),
65
>
}
66
>
}
67
69
>
}
70
72
>
close(tm.closeC)
73
>
}
74
76
>
tm.rateLimiter.RecycleToken()
77
>
}
78
79
// Offer offers a task to a potential consumer (poller)
80
// If the task is successfully matched with a consumer, this
81
// method will return true and no error. If the task is matched
82
// but consumer returned error, then this method will return
83
// true and error message. This method should not be used for query
84
// task. This method should ONLY be used for sync match.
85
//
86
// When a local poller is not available and forwarding to a parent
87
// task queue partition is possible, this method will attempt forwarding
88
// to the parent partition.
89
//
90
// Cases when this method will block:
91
//
92
// Ratelimit:
93
// When a ratelimit token is not available, this method might block
94
// waiting for a token until the provided context timeout. Rate limits are
95
// not enforced for forwarded tasks from child partition.
96
//
97
// Forwarded tasks that originated from db backlog:
98
// When this method is called with a task that is forwarded from a
99
// remote partition and if (1) this task queue is root (2) task
100
// was from db backlog - this method will block until context timeout
101
// trying to match with a poller. The caller is expected to set the
102
// correct context timeout.
103
//
104
// returns error when:
105
// - ratelimit is exceeded (does not apply to query task)
106
// - context deadline is exceeded
107
// - task is matched and consumer returns error in response channel
108
>
func (tm *TaskMatcher) Offer(ctx context.Context, task *internalTask) (bool, error) {
matcher.go ×1
109
>
if !tm.isBacklogNegligible() {
110
>
// To ensure better dispatch ordering, we block sync match when a significant backlog is present.
matcher.go ×1
111
>
// Note that this check does not make a noticeable difference for history tasks, as they do not wait for a
112
>
// poller to become available. In presence of a backlog the chance of a poller being available when sync match
113
>
// request comes is almost zero.
114
>
// This check is mostly effective for the sync match requests that come from child partitions for spooled tasks.
115
>
return false, nil
116
>
}
117
121
>
return false, err
122
>
}
123
// because we waited on the rate limiter to offer this task,
124
// attach the rate limiter's RecycleToken func to the task
125
// so that if the task is later determined to be invalid,
126
// we can recycle the token it used.
128
}
129
132
>
if task.responseC != nil {
133
>
// if there is a response channel, block until resp is received
134
>
// and return error if the response contains error
135
>
err := <-task.responseC
136
>
137
>
if err.startErr == nil && !task.isForwarded() {
139
>
}
141
}
142
return false, nil
144
>
// no poller waiting for tasks, try forwarding this task to the
145
>
// root partition if possible
146
>
select {
148
>
if err := tm.fwdr.ForwardTask(ctx, task); err == nil {
150
>
token.release()
151
>
if !task.isForwarded() {
152
// if there are multiple forwarding hops, only the initial source partition emits this metric
153
tm.emitDispatchLatency(task, true)
154
}
156
}
159
>
if !tm.isForwardingAllowed() && // we are the root partition and forwarding is not possible
160
>
task.source == enumsspb.TASK_SOURCE_DB_BACKLOG && // task was from backlog (stored in db)
161
>
task.isForwarded() { // task came from a child partition
163
>
// to match with a poller until ctx timeout
164
>
return tm.offerOrTimeout(ctx, task)
165
>
}
166
}
167
169
}
170
}
171
172
>
func (tm *TaskMatcher) offerOrTimeout(ctx context.Context, task *internalTask) (bool, error) {
matcher.go ×3
173
>
select {
174
>
case tm.taskC <- task: // poller picked up the task
175
>
if task.responseC != nil {
176
>
select {
177
>
case err := <-task.responseC:
178
>
return true, err.startErr
179
case <-ctx.Done():
180
return false, nil
181
}
182
}
183
return false, nil
184
case <-ctx.Done():
185
return false, nil
186
}
187
}
188
189
func syncOfferTask[T any](
190
ctx context.Context,
191
tm *TaskMatcher,
192
task *internalTask,
193
taskChan chan *internalTask,
194
forwardFunc func(context.Context, *internalTask) (T, error),
195
returnNoPollerErr bool,
197
>
var t T
198
>
select {
200
>
<-task.responseC
201
>
return t, nil
203
}
204
206
>
var noPollerC <-chan time.Time
207
>
208
>
for {
209
>
if returnNoPollerErr {
211
>
if deadline, ok := ctx.Deadline(); ok && fwdrTokenC == nil {
213
>
// without having started the workers.
214
>
noPollerTimeout := time.Until(deadline) - returnEmptyTaskTimeBudget
215
>
t := time.NewTimer(noPollerTimeout)
216
>
noPollerC = t.C
217
>
defer t.Stop()
218
>
}
219
}
220
223
>
<-task.responseC
224
>
return t, nil
226
>
resp, err := forwardFunc(ctx, task)
227
>
token.release()
228
>
if err == nil {
230
>
}
232
>
// if we are rate limited, try only local match for the remainder of the context timeout
matcher.go ×2
233
>
// left
234
>
fwdrTokenC = nil
235
>
continue
236
}
239
>
// only error if there has not been a recent poller. Otherwise, let it wait for the remaining time
240
>
// hopping for a match, or ultimately returning the default CDE error.
241
>
if tm.timeSinceLastPoll() > tm.config.QueryPollerUnavailableWindow() {
243
>
}
246
>
return t, ctx.Err()
247
}
248
}
249
}
250
251
// OfferQuery will either match task to local poller or will forward query task.
252
// Local match is always attempted before forwarding is attempted. If local match occurs
253
// response and error are both nil, if forwarding occurs then response or error is returned.
254
>
func (tm *TaskMatcher) OfferQuery(ctx context.Context, task *internalTask) (*matchingservice.QueryWorkflowResponse, error) {
matcher.go ×1
255
>
return syncOfferTask(ctx, tm, task, tm.queryTaskC, tm.fwdr.ForwardQueryTask, true)
256
>
}
257
258
// OfferNexusTask either matchs a task to a local poller or forwards it if no local pollers available.
259
// Local match is always attempted before forwarding. If local match occurs response and error are both nil, if
260
// forwarding occurs then response or error is returned.
261
>
func (tm *TaskMatcher) OfferNexusTask(ctx context.Context, task *internalTask) (*matchingservice.DispatchNexusTaskResponse, error) {
matcher.go ×1
262
>
return syncOfferTask(ctx, tm, task, tm.taskC, tm.fwdr.ForwardNexusTask, false)
263
>
}
264
265
// MustOffer blocks until a consumer is found to handle this task
266
// Returns error only when context is canceled or the ratelimit is set to zero (allow nothing)
267
// The passed in context MUST NOT have a deadline associated with it
268
// Note that calling MustOffer is the only way that matcher knows there are spooled tasks in the
269
// backlog, in absence of a pending MustOffer call, the forwarding logic assumes that backlog is empty.
270
>
func (tm *TaskMatcher) MustOffer(ctx context.Context, task *internalTask, interruptCh <-chan struct{}) error {
matcher.go ×3
271
>
tm.registerBacklogTask(task)
272
>
defer tm.unregisterBacklogTask(task)
273
>
274
>
if err := tm.rateLimiter.Wait(ctx); err != nil {
276
>
}
277
278
// because we waited on the rate limiter to offer this task,
279
// attach the rate limiter's RecycleToken func to the task
280
// so that if the task is later determined to be invalid,
281
// we can recycle the token it used.
283
>
284
>
// attempt a match with local poller first. When that
285
>
// doesn't succeed, try both local match and remote match
286
>
select {
288
>
tm.emitDispatchLatency(task, false)
289
>
return nil
291
>
return ctx.Err()
293
}
294
296
>
defer func() {
299
>
}
300
}()
301
303
>
for {
304
>
fwdTokenC := tm.fwdrAddReqTokenC()
305
>
reconsiderFwdTimer = nil
306
>
var reconsiderFwdTimerC <-chan time.Time
307
>
if fwdTokenC != nil && !tm.isBacklogNegligible() {
309
>
// root and leaf partitions are treated equally and can process their
310
>
// backlog at the same rate. Stopping task forwarding, prevent poll
311
>
// forwarding as well (in presence of a backlog). This ensures all partitions
312
>
// receive polls and tasks at the same rate.
313
>
314
>
// Exception: we allow forward if this partition has not got any polls
315
>
// recently. This is helpful when there are very few pollers and they
316
>
// and they are all stuck in the wrong (root) partition. (Note that since
317
>
// frontend balanced the number of pending pollers per partition this only
318
>
// becomes an issue when the pollers are fewer than the partitions)
319
>
lp := tm.timeSinceLastPoll()
320
>
maxWaitForLocalPoller := tm.config.MaxWaitForPollerBeforeFwd()
321
>
if lp < maxWaitForLocalPoller {
322
>
fwdTokenC = nil
323
>
reconsiderFwdTimer = time.NewTimer(maxWaitForLocalPoller - lp)
324
>
reconsiderFwdTimerC = reconsiderFwdTimer.C
325
>
}
326
}
327
330
>
tm.emitDispatchLatency(task, false)
331
>
return nil
333
>
childCtx, cancel := context.WithTimeout(ctx, time.Second*2)
334
>
err := tm.fwdr.ForwardTask(childCtx, task)
335
>
token.release()
336
>
if err != nil {
338
>
// forwarder returns error only when the call is rate limited. To
339
>
// avoid a busy loop on such rate limiting events, we only attempt to make
340
>
// the next forwarded call after this childCtx expires. Till then, we block
341
>
// hoping for a local poller match
342
>
select {
343
case tm.taskC <- task:
344
cancel()
345
tm.emitDispatchLatency(task, false)
346
return nil
349
>
cancel()
350
>
return ctx.Err()
351
case <-interruptCh:
352
cancel()
353
return errInterrupted
354
}
356
>
continue forLoop
357
}
359
>
// at this point, we forwarded the task to a parent partition which
360
>
// in turn dispatched the task to a poller, because there was no error.
361
>
// Make sure we delete the task from the database.
362
>
task.finish(taskFinishResult{consumedToken: true})
363
>
tm.emitDispatchLatency(task, true)
364
>
return nil
366
>
return ctx.Err()
368
>
continue forLoop
369
case <-interruptCh:
370
return errInterrupted
371
}
372
}
373
}
374
375
>
func (tm *TaskMatcher) emitDispatchLatency(task *internalTask, forwarded bool) {
matcher.go ×1
376
>
if tm.config.EmitTaskDispatchLatencyAtPoll() {
377
>
return // metric will be emitted at poll response
378
>
}
379
if task.event.Data.CreateTime == nil {
380
return // should not happen but for safety
381
}
382
383
metrics.TaskDispatchLatencyPerTaskQueue.With(tm.metricsHandler).Record(
384
time.Since(timestamp.TimeValue(task.event.Data.CreateTime)),
385
metrics.StringTag("source", task.source.String()),
386
metrics.ForwardedTag(forwarded),
387
metrics.StringTag(metrics.TaskPriorityTagName, ""),
388
)
389
}
390
391
// Poll blocks until a task is found or context deadline is exceeded
392
// On success, the returned task could be a query task or a regular task
393
// Returns errNoTasks when context deadline is exceeded
394
>
func (tm *TaskMatcher) Poll(ctx context.Context, pollMetadata *pollMetadata) (*internalTask, error) {
matcher.go ×1
395
>
task, _, err := tm.poll(ctx, pollMetadata, false)
396
>
return task, err
397
>
}
398
399
// PollForQuery blocks until a *query* task is found or context deadline is exceeded
400
// Returns errNoTasks when context deadline is exceeded
401
>
func (tm *TaskMatcher) PollForQuery(ctx context.Context, pollMetadata *pollMetadata) (*internalTask, error) {
matcher.go ×2
402
>
task, _, err := tm.poll(ctx, pollMetadata, true)
403
>
return task, err
404
>
}
405
407
>
// unused in old matcher
408
>
}
409
410
func (tm *TaskMatcher) poll(
411
ctx context.Context, pollMetadata *pollMetadata, queryOnly bool,
413
>
taskC, queryTaskC := tm.taskC, tm.queryTaskC
414
>
if queryOnly {
416
>
}
417
419
>
tm.lastPoller.Store(start.UnixNano())
420
>
421
>
defer func() {
422
>
if pollMetadata.forwardedFrom == "" {
423
>
// Only recording for original polls
424
>
var pollResult string
425
>
if err == nil {
429
>
} else {
430
pollResult = "failed"
431
}
433
>
time.Since(start),
434
>
metrics.ForwardedTag(forwardedPoll),
435
>
metrics.StringTag(metrics.TaskPriorityTagName, ""),
436
>
metrics.PollResultTag(pollResult),
437
>
)
438
}
439
441
>
tm.emitForwardedSourceStats(task.isForwarded(), pollMetadata.forwardedFrom, forwardedPoll)
matcher.go ×3
442
>
}
443
}()
444
445
// We want to effectively do a prioritized select, but Go select is random
446
// if multiple cases are ready, so split into multiple selects.
447
// The priority order is:
448
// 1. ctx.Done or tm.closeC
449
// 2. taskC and queryTaskC
450
// 3. forwarding
451
// 4. block looking locally for remainder of context lifetime
452
// To correctly handle priorities and allow any case to succeed, all select
453
// statements except for the last one must be non-blocking, and the last one
454
// must include all the previous cases.
455
456
// 1. ctx.Done
459
>
metrics.PollTimeoutPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
460
>
return nil, false, errNoTasks
462
>
return nil, false, errNoTasks
464
}
465
466
// 2. taskC and queryTaskC
469
>
if task.responseC != nil {
470
>
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
matcher.go ×3
471
>
}
473
>
return task, false, nil
474
case task := <-queryTaskC:
475
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
476
metrics.PollSuccessPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
477
return task, false, nil
479
}
480
481
// From here on the goroutine will block on taskC (in step 3 or 4), so it
482
// is ready to receive a sync-matched task.
484
>
defer tm.waitingPollerCount.Add(-1)
485
>
486
>
if tm.isBacklogNegligible() {
488
>
// We don't forward pollers if there is a non-negligible backlog in this partition.
489
>
select {
491
>
metrics.PollTimeoutPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
492
>
return nil, false, errNoTasks
494
>
return nil, false, errNoTasks
496
>
if task.responseC != nil {
497
>
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
matcher.go ×1
498
>
}
500
>
return task, false, nil
502
>
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
503
>
metrics.PollSuccessPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
504
>
return task, false, nil
506
>
// Arrange to cancel this request if closeC is closed
507
>
fwdCtx, cancel := contextWithCancelOnChannelClose(ctx, tm.closeC)
508
>
task, err := tm.fwdr.ForwardPoll(fwdCtx, pollMetadata)
509
>
cancel()
510
>
token.release()
511
>
if err == nil {
513
>
}
514
}
515
}
516
517
// 4. blocking local poll
520
>
metrics.PollTimeoutPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
521
>
return nil, false, errNoTasks
522
case <-tm.closeC:
523
return nil, false, errNoTasks
525
>
if task.responseC != nil {
526
>
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
matcher.go ×2
527
>
}
529
>
return task, false, nil
530
case task := <-queryTaskC:
531
metrics.PollSuccessWithSyncPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
532
metrics.PollSuccessPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
533
return task, false, nil
534
}
535
}
536
538
>
if tm.fwdr == nil {
540
>
}
542
}
543
545
>
if tm.fwdr == nil {
547
>
}
549
}
550
552
>
return tm.fwdr != nil
553
>
}
554
555
// isBacklogNegligible returns true of the age of backlog is less than the threshold. Note that this relies on
556
// MustOffer being called when there is a backlog, otherwise we'd not know.
558
>
return tm.getBacklogAge() < tm.config.BacklogNegligibleAge()
559
>
}
560
562
>
if task.event.Data.CreateTime == nil {
564
>
}
565
567
>
defer tm.backlogTasksLock.Unlock()
568
>
569
>
ts := timestamp.TimeValue(task.event.Data.CreateTime).UnixNano()
570
>
tm.backlogTasksCreateTime[ts] += 1
571
}
572
574
>
if task.event.Data.CreateTime == nil {
576
>
}
577
579
>
defer tm.backlogTasksLock.Unlock()
580
>
581
>
ts := timestamp.TimeValue(task.event.Data.CreateTime).UnixNano()
582
>
counter := tm.backlogTasksCreateTime[ts]
583
>
if counter == 1 {
584
>
delete(tm.backlogTasksCreateTime, ts)
585
>
} else {
587
>
}
588
}
589
590
// getBacklogAge is the latest age across all backlogs re-directing to this matcher; may momentarily
591
// be 0 cause of race conditions when no reader pushes a task into the matcher at this moment
593
>
tm.backlogTasksLock.Lock()
594
>
defer tm.backlogTasksLock.Unlock()
595
>
596
>
if len(tm.backlogTasksCreateTime) == 0 {
598
>
}
599
601
>
for createTime := range tm.backlogTasksCreateTime {
602
>
oldest = min(oldest, createTime)
603
>
}
604
606
}
607
608
func (tm *TaskMatcher) emitForwardedSourceStats(
609
isTaskForwarded bool,
610
pollForwardedSource string,
611
forwardedPoll bool,
613
>
if forwardedPoll {
614
>
// This means we forwarded the poll to another partition. Skipping this to prevent duplicate emits.
matcher.go ×2
615
>
// Only the partition in which the match happened should emit this metric.
616
>
return
617
>
}
618
620
>
switch {
621
case isTaskForwarded && isPollForwarded:
622
metrics.RemoteToRemoteMatchPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
624
>
metrics.RemoteToLocalMatchPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
625
case isPollForwarded:
626
metrics.LocalToRemoteMatchPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
628
>
metrics.LocalToLocalMatchPerTaskQueueCounter.With(tm.metricsHandler).Record(1)
629
}
630
}
631
633
>
return time.Since(time.Unix(0, tm.lastPoller.Load()))
634
>
}
635
636
// HasWaitingPoller returns true if it there's a poller ready and waiting
637
// this is mostly useful in testing to avoid test races on setup
639
>
return tm.waitingPollerCount.Load() > 0
640
>
}
641
642
// contextWithCancelOnChannelClose returns a child Context and CancelFunc just like
643
// context.WithCancel, but additionally propagates cancellation from another channel (besides
644
// the parent's cancellation channel).
645
>
func contextWithCancelOnChannelClose(parent context.Context, closeC <-chan struct{}) (context.Context, context.CancelFunc) {
matcher.go ×3
646
>
ctx, cancel := context.WithCancel(parent)
647
>
go func() {
648
>
select {
650
>
cancel()
652
}
653
}()
655
}