Vihren Atlas

matching_engine.go ×8

Code frontier

unlabeled · c_793d67091535

7 tests · 3147 LOC · 145 files · introduces 0 tests · 77 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
11 ranges77 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
492 ranges3147 lines · 145 files · Browse complete extent
All tests (intent)
7 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.

2 files ranked by introduced lines: 77 introduced LOC across 11 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/matching_engine.go 63 introduced LOC · 8 ranges

Open complete file

1233 }
1234 if e.config.EnableMatchingFanOutForPollCancellation(ns.String()) {
1235 > // TODO: Remove the IsRoot/Sticky guard after EnableMatchingFanOutForPollCancellation is matching_engine.go
1236 > // fully rolled out. This check is only needed during the transition since the legacy
1237 > // frontend fan-out path may send non-root partitions to this handler.
1238 > if partition.IsRoot() && partition.Kind() != enumspb.TASK_QUEUE_KIND_STICKY {
1239 > return e.cancelOutstandingWorkerPollsForAllPartitions(ctx, request, partition)
1240 > }
1241 }
1242 // TODO: Delete this code path after EnableMatchingFanOutForPollCancellation is rolled out.
1256 request *matchingservice.CancelOutstandingWorkerPollsRequest,
1257 rootPartition tqid.Partition,
1258 > ) (*matchingservice.CancelOutstandingWorkerPollsResponse, error) { matching_engine.go
1259 > rootPM, _, err := e.getTaskQueuePartitionManager(ctx, rootPartition, false, loadCauseOtherWrite)
1260 > if err != nil {
1261 return nil, err
1262 }
1263 > if rootPM == nil { matching_engine.go
1264 // Root not loaded means no pending polls anywhere — child partitions loading
1265 // triggers root to load via user data fetch chain.
1272 return &matchingservice.CancelOutstandingWorkerPollsResponse{}, nil
1273 }
1274 > cfg := rootPM.GetConfig() matching_engine.go
1275 > // TODO(dynamic partitioning): get real num read partitions from the partition manager.
1276 > numPartitions := cfg.NumReadPartitions()
1277 >
1278 > e.logger.Debug("Initiating fan-out for worker poll cancellation",
1279 > tag.WorkflowNamespaceID(request.GetNamespaceId()),
1280 > tag.WorkflowTaskQueueName(rootPartition.TaskQueue().Name()),
1281 > tag.WorkflowTaskQueueType(request.GetTaskQueueType()),
1282 > tag.NewStringTag("worker-instance-key", request.GetWorkerInstanceKey()),
1283 > tag.NewInt32("partition-count", int32(numPartitions)),
1284 > )
1285 >
1286 > workers := []*matchingservice.CancelOutstandingWorkerPollsPartitionRequest_WorkerEntry{{
1287 > WorkerInstanceKey: request.GetWorkerInstanceKey(),
1288 > WorkerIdentity: request.GetWorkerIdentity(),
1289 > }}
1290 >
1291 > // Group partitions by destination host. When Route() is unavailable or fails, each
1292 > // unroutable partition gets a synthetic key so it's sent as an individual RPC.
1293 > routingClient, _ := e.matchingRawClient.(matching.RoutingClient) //nolint:revive // unchecked-type-assertion: nil is the desired zero value
1294 > self := e.hostInfoProvider.HostInfo().Identity()
1295 > tq := rootPartition.TaskQueue()
1296 > partitionsByTarget := make(map[string][]*tqid.NormalPartition, numPartitions)
1297 >
1298 > for i := range numPartitions {
1299 > partition := tq.NormalPartition(i)
1300 > target := ""
1301 > if routingClient != nil {
1302 h, err := routingClient.Route(partition)
1303 if err != nil {
1309 }
1310 }
1311 > if target == "" { matching_engine.go
1312 target = fmt.Sprintf("_unroutable_%d", i)
1313 }
1314 > partitionsByTarget[target] = append(partitionsByTarget[target], partition) matching_engine.go
1315 }
1316
1317 // Process each target: local via direct call, remote via RPC.
1318 > var totalCancelled atomic.Int32 matching_engine.go
1319 > var wg sync.WaitGroup
1320 >
1321 > for target, partitions := range partitionsByTarget {
1322 > partitionProtos := make([]*taskqueuespb.TaskQueuePartition, len(partitions))
1323 > for i, np := range partitions {
1324 > partitionProtos[i] = &taskqueuespb.TaskQueuePartition{
1325 > TaskQueue: np.TaskQueue().Name(),
1326 > TaskQueueType: np.TaskType(),
1327 > PartitionId: &taskqueuespb.TaskQueuePartition_NormalPartitionId{NormalPartitionId: int32(np.PartitionId())},
1328 > }
1329 > }
1330 > req := &matchingservice.CancelOutstandingWorkerPollsPartitionRequest{
1331 > NamespaceId: request.GetNamespaceId(),
1332 > TaskQueuePartition: partitionProtos[0], // routing key
1333 > Partitions: partitionProtos,
1334 > Workers: workers,
1335 > }
1336 > if target == self {
1337 resp, err := e.CancelOutstandingWorkerPollsPartition(ctx, req)
1338 if err != nil {
1358 }
1359
1360 > wg.Wait() matching_engine.go
1361 > return &matchingservice.CancelOutstandingWorkerPollsResponse{
1362 > CancelledCount: totalCancelled.Load(),
1363 > }, nil
1364 }
1365
go.temporal.io/server/service/matching/task_queue_partition_manager_mock.go 14 introduced LOC · 3 ranges

Open complete file

150
151 // GetConfig mocks base method.
152 > func (m *MocktaskQueuePartitionManager) GetConfig() *taskQueueConfig { task_queue_partition_manager_mock.go
153 > m.ctrl.T.Helper()
154 > ret := m.ctrl.Call(m, "GetConfig")
155 > ret0, _ := ret[0].(*taskQueueConfig)
156 > return ret0
157 > }
158
159 // GetConfig indicates an expected call of GetConfig.
160 > func (mr *MocktaskQueuePartitionManagerMockRecorder) GetConfig() *gomock.Call { task_queue_partition_manager_mock.go
161 > mr.mock.ctrl.T.Helper()
162 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetConfig", reflect.TypeOf((*MocktaskQueuePartitionManager)(nil).GetConfig))
163 > }
164
165 // GetRateLimitManager mocks base method.
351
352 // RemovePoller indicates an expected call of RemovePoller.
353 > func (mr *MocktaskQueuePartitionManagerMockRecorder) RemovePoller(identity any) *gomock.Call { task_queue_partition_manager_mock.go
354 > mr.mock.ctrl.T.Helper()
355 > return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RemovePoller", reflect.TypeOf((*MocktaskQueuePartitionManager)(nil).RemovePoller), identity)
356 > }
357
358 // Start mocks base method.

Direct hierarchy

Parents

Children