Vihren Atlas

task_queue_partition_manager.go ×12

Code frontier

unlabeled · c_71c364c44c05

232 tests · 4251 LOC · 175 files · introduces 0 tests · 52 LOC · 3 files

Introduces — evidence that enters the hierarchy at this concept

Code
16 ranges52 lines · 3 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
764 ranges4251 lines · 175 files · Browse complete extent
All tests (intent)
232 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.

3 files ranked by introduced lines: 52 introduced LOC across 16 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/task_queue_partition_manager.go 38 introduced LOC · 12 ranges

Open complete file

524 }
525
526 > func (pm *taskQueuePartitionManagerImpl) autoEnableIfNeeded(ctx context.Context, params addTaskParams) { task_queue_partition_manager.go
527 > if pm.fairnessState != enumsspb.FAIRNESS_STATE_UNSPECIFIED {
528 return
529 }
530 > if params.taskInfo.Priority.GetFairnessKey() == "" { task_queue_partition_manager.go
531 if params.taskInfo.Priority.GetPriorityKey() == int32(0) {
532 return
558 ctx context.Context,
559 params addTaskParams,
560 > ) (buildId string, syncMatched bool, err error) { task_queue_partition_manager.go
561 > defer pm.sendPartitionCountTrailer(ctx)
562 > if err := pm.checkPartitionCounts(ctx, true); err != nil {
563 return "", false, err
564 }
565 > if params.forwardInfo == nil { task_queue_partition_manager.go
566 pm.signalPartitionScaler()
567 }
568
569 > var spoolQueue, syncMatchQueue physicalTaskQueueManager task_queue_partition_manager.go
570 > directive := params.taskInfo.GetVersionDirective()
571 >
572 > pm.autoEnableIfNeeded(ctx, params)
573 > // spoolQueue will be nil iff task is forwarded.
574 > reredirectTask:
575 > spoolQueue, syncMatchQueue, _, taskDispatchRevisionNumber, targetVersion, err := pm.getPhysicalQueuesForAdd(ctx, directive, params.forwardInfo, params.taskInfo.GetRunId(), params.taskInfo.GetWorkflowId(), false)
576 > if err != nil {
577 return "", false, err
578 }
579
580 > syncMatchTask := newInternalTaskForSyncMatch(params.taskInfo, params.forwardInfo, taskDispatchRevisionNumber, targetVersion) task_queue_partition_manager.go
581 > pm.config.setDefaultPriority(syncMatchTask)
582 > if spoolQueue != nil && spoolQueue.QueueKey().Version().BuildId() != syncMatchQueue.QueueKey().Version().BuildId() {
583 // Task is not forwarded and build ID is different on the two queues -> redirect rule is being applied.
584 // Set redirectInfo in the task as it will be needed if we have to forward the task.
588 }
589
590 > dbq := pm.defaultQueue() task_queue_partition_manager.go
591 > if dbq == nil {
592 return "", false, errDefaultQueueNotInit
593 }
594 > if dbq != syncMatchQueue { task_queue_partition_manager.go
595 // default queue should stay alive even if requests go to other queues
596 dbq.MarkAlive()
597 }
598
599 > if pm.partition.IsRoot() { task_queue_partition_manager.go
600 // Only emit the no-recent-poller metric if BOTH conditions are met:
601 // 1. Partition has been loaded for more than noPollerThreshold (2 minutes)
607 }
608
609 > isActive, err := pm.isActiveInCluster() task_queue_partition_manager.go
610 > if err != nil {
611 return "", false, err
612 }
613
614 > behavior := directive.GetBehavior() task_queue_partition_manager.go
615 > forwarded := params.forwardInfo != nil
616 >
617 > var outcome syncMatchOutcome
618 > if isActive {
619 > outcome, err = syncMatchQueue.TrySyncMatch(ctx, syncMatchTask)
620 > syncMatched = outcome == syncMatchSuccess
621 > if syncMatched && !pm.shouldBacklogSyncMatchTaskOnError(err) {
622 // Only fire hooks for non-forwarded tasks. Forwarded tasks already had hooks fired
623 // on the child partition that originally received the task.
718 }
719
720 > func (pm *taskQueuePartitionManagerImpl) isActiveInCluster() (bool, error) { task_queue_partition_manager.go
721 > ns, err := pm.engine.namespaceRegistry.GetNamespaceByID(pm.ns.ID())
722 > if err == nil {
723 > //nolint:forbidigo // partition manager is namespace-scoped
724 > return ns.ActiveInCluster(pm.engine.clusterMeta.GetCurrentClusterName()), nil
725 > }
726 return false, err
727 }
go.temporal.io/server/service/matching/physical_task_queue_manager.go 11 introduced LOC · 3 ranges

Open complete file

471 }
472
473 > func (c *physicalTaskQueueManagerImpl) RecordTaskAdd(result string, forwarded bool, behavior enumspb.VersioningBehavior) { physical_task_queue_manager.go
474 > c.metricsHandler.Counter(metrics.TasksAddedCounter.Name()).Record(
475 > 1,
476 > metrics.TaskAddResultTag(result),
477 > metrics.ForwardedTag(forwarded),
478 > metrics.VersioningBehaviorTag(behavior),
479 > )
480 > }
481
482 // PollTask blocks waiting for a task.
720 }
721
722 > func (c *physicalTaskQueueManagerImpl) TrySyncMatch(ctx context.Context, task *internalTask) (syncMatchOutcome, error) { physical_task_queue_manager.go
723 > if !task.isForwarded() {
724 // request sent by history service
725 c.liveness.markAlive()
go.temporal.io/server/common/metrics/tags.go 3 introduced LOC · 1 range

Open complete file

331 )
332
333 > func TaskAddResultTag(result string) Tag { tags.go
334 > return Tag{Key: taskAddResult, Value: result}
335 > }
336
337 const (

Direct hierarchy

Parents

Children