Vihren Atlas

pri_matcher.go ×8

Code frontier

unlabeled · c_0b5a6ab63667

5 tests · 40719 LOC · 747 files · introduces 0 tests · 55 LOC · 5 files

Introduces — evidence that enters the hierarchy at this concept

Code
17 ranges55 lines · 5 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
9479 ranges40719 lines · 747 files · Browse complete extent
All tests (intent)
5 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

5 files ranked by introduced lines: 55 introduced LOC across 17 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/pri_matcher.go 27 introduced LOC · 8 ranges

Open complete file

178 return // task queue closing
179 }
180 > if !softassert.That(tm.logger, res.task != nil, "expected a task from match") { pri_matcher.go
181 continue
182 }
183
184 > skipLimiter, err = tm.forwardTask(res.task) pri_matcher.go
185 >
186 > // backoff on resource exhausted errors
187 > if common.IsResourceExhausted(err) {
188 util.InterruptibleSleep(tm.tqCtx, retrier.NextBackOff(err))
189 > } else { pri_matcher.go
190 > retrier.Reset()
191 > }
192 }
193 }
194
195 > func (tm *priTaskMatcher) forwardTask(task *internalTask) (bool, error) { pri_matcher.go
196 > var ctx context.Context
197 > var cancel context.CancelFunc
198 > if task.forwardCtx != nil {
199 > // Use sync match context if we have it (for deadline, headers, etc.)
200 > // TODO(pri): does it make sense to subtract 1s from the context deadline here?
201 > // Also arrange for this to be canceled on tqCtx closing.
202 > ctx, cancel = context.WithCancel(task.forwardCtx)
203 > stop := context.AfterFunc(tm.tqCtx, cancel)
204 > defer cancel()
205 > defer stop()
206 > } else {
207 // Task is from local backlog.
208
226 }
227
228 > if task.isQuery() { pri_matcher.go
229 res, err := tm.fwdr.ForwardQueryTask(ctx, task)
230 task.finishForward(res, err, true)
232 }
233
234 > if task.isNexus() { pri_matcher.go
235 res, err := tm.fwdr.ForwardNexusTask(ctx, task)
236 task.finishForward(res, err, true)
239
240 // normal wf/activity task
241 > err := tm.fwdr.ForwardTask(ctx, task) pri_matcher.go
242 > task.finishForward(nil, err, true)
243 >
244 > return false, err
245 }
246
393 }
394 if res.forwarded {
395 > if res.forwardErr == nil { pri_matcher.go
396 // task was remotely sync matched on the parent partition
397 tm.emitDispatchLatency(task, true)
go.temporal.io/server/service/matching/pri_forwarder.go 16 introduced LOC · 5 ranges

Open complete file

71 }
72
73 > var expirationDuration *durationpb.Duration pri_forwarder.go
74 > var expirationTime time.Time
75 > if task.event.Data.ExpiryTime != nil {
76 expirationTime = task.event.Data.ExpiryTime.AsTime()
77 remaining := time.Until(expirationTime)
83
84 // nolint:exhaustive // there's a default clause
85 > switch f.partition.TaskType() { pri_forwarder.go
86 case enumspb.TASK_QUEUE_TYPE_WORKFLOW:
87 _, err = f.client.AddWorkflowTask(
125 }
126
127 > return err pri_forwarder.go
128 }
129
130 > func (f *priForwarder) getForwardInfo(task *internalTask) *taskqueuespb.TaskForwardInfo { pri_forwarder.go
131 > if task.isForwarded() {
132 // task is already forwarded from a child partition, only overwrite SourcePartition
133 clone := common.CloneProto(task.forwardInfo)
136 }
137 // task is forwarded for the first time
138 > return &taskqueuespb.TaskForwardInfo{ pri_forwarder.go
139 > CreateTime: task.getCreateTime(),
140 > TaskSource: task.source,
141 > SourcePartition: f.partition.RpcName(),
142 > OriginPartition: f.partition.RpcName(),
143 > DispatchBuildId: f.queue.Version().BuildId(),
144 > DispatchVersionSet: f.queue.Version().VersionSet(),
145 > RedirectInfo: task.redirectInfo,
146 > }
147 }
148
go.temporal.io/server/api/taskqueue/v1/message.pb.go 4 introduced LOC · 1 range

Open complete file

756 }
757
758 > func (x *TaskForwardInfo) GetOriginPartition() string { message.pb.go
759 > if x != nil {
760 > return x.OriginPartition
761 > }
762 return ""
763 }
go.temporal.io/server/service/matching/matching_engine.go 4 introduced LOC · 1 range

Open complete file

3107 originPartition := partition
3108 if task.isForwarded() && task.forwardInfo.GetOriginPartition() != "" {
3109 > o, err := tqid.NormalPartitionFromRpcName(task.forwardInfo.GetOriginPartition(), namespaceID, taskType) matching_engine.go
3110 > if err == nil {
3111 > originPartition = o
3112 > } // else ignore the error and use the current partition
3113 }
3114
go.temporal.io/server/service/matching/task.go 4 introduced LOC · 2 ranges

Open complete file

239
240 // isNexus returns true if the underlying task is a nexus task
241 > func (task *internalTask) isNexus() bool { task.go
242 > return task.nexus != nil
243 > }
244
245 // isStarted is true when this task is already marked as started
260 func (task *internalTask) getCreateTime() *timestamppb.Timestamp {
261 if task.forwardInfo.GetCreateTime() != nil {
262 > return task.forwardInfo.GetCreateTime() task.go
263 } else if task.event != nil {
264 return task.event.Data.GetCreateTime()

Direct hierarchy

Parents

Children