599
func (tm *priTaskMatcher) poll(
600
ctx context.Context, pollMetadata *pollMetadata, queryOnly bool,
602
>
start := time.Now()
603
>
pollWasForwarded := false
604
>
var priority int32
605
>
pollResult := "failed"
606
>
607
>
defer func() {
608
>
// TODO(pri): can we consolidate all the metrics code below?
609
>
if pollMetadata.forwardedFrom == "" {
610
>
// Only recording for original polls (i.e. on child if forwarded)
611
>
metrics.PollLatencyPerTaskQueue.With(tm.metricsHandler).Record(
612
>
time.Since(start),
613
>
metrics.ForwardedTag(pollWasForwarded),
614
>
metrics.MatchingTaskPriorityTag(priority),
615
>
metrics.PollResultTag(pollResult),
616
>
)
617
>
}
618
}()
619
621
>
startTime: start,
622
>
queryOnly: queryOnly,
623
>
forwardCtx: ctx,
624
>
pollMetadata: pollMetadata,
625
>
}
626
>
627
>
var res *matchResult
628
>
if pollMetadata.conditions.GetNoWait() {
629
res = tm.data.MatchPollerImmediately(poller)
631
>
ctxs := []context.Context{ctx, tm.tqCtx}
632
>
res = tm.data.EnqueuePollerAndWait(ctxs, poller)
633
>
}
634
636
pollResult = "timeout"
637
return nil, errNoTasks // only possible for MatchPollerImmediately
639
if res.ctxErrIdx == 0 {
640
metrics.PollTimeoutPerTaskQueueCounter.With(tm.metricsHandler).Record(1)