Vihren Atlas

task_queue_partition_manager.go ×19

Code frontier

unlabeled · c_a97672737630

168 tests · 4323 LOC · 174 files · introduces 0 tests · 138 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
23 ranges138 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
771 ranges4323 lines · 174 files · Browse complete extent
All tests (intent)
168 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: 138 introduced LOC across 23 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/matching/task_queue_partition_manager.go 123 introduced LOC · 19 ranges

Open complete file

1302 buildIds map[string]bool,
1303 includeAllActive, reportStats, reportPollers, internalTaskQueueStatus, skipMarkAlive bool,
1304 > ) (*matchingservice.DescribeTaskQueuePartitionResponse, error) { task_queue_partition_manager.go
1305 > pm.versionedQueuesLock.RLock()
1306 >
1307 > versions := make(map[PhysicalTaskQueueVersion]bool)
1308 >
1309 > // Active means that the physical queue for that version is loaded.
1310 > // An empty string refers to the unversioned queue, which is always loaded.
1311 > // In the future, active will mean that the physical queue for that version has had a task added recently or a recent poller.
1312 > if includeAllActive {
1313 for k := range pm.versionedQueues {
1314 versions[k] = true
1352 }
1353
1354 > pm.versionedQueuesLock.RUnlock() task_queue_partition_manager.go
1355 >
1356 > var unversionedStatsByPriority map[int32]*taskqueuepb.TaskQueueStats
1357 > var currentVersion *deploymentspb.WorkerDeploymentVersion
1358 > var rampingVersion *deploymentspb.WorkerDeploymentVersion
1359 > var rampPercentage float32
1360 > var currentExists bool
1361 > var rampingExists bool
1362 > var isRamping bool
1363 > var unversionedCurrentShareByPriority map[int32]*taskqueuepb.TaskQueueStats
1364 > var unversionedRampingShareByPriority map[int32]*taskqueuepb.TaskQueueStats
1365 >
1366 > if reportStats {
1367 > // Consider the default/unversioned queue. For current/ramping deployment versions, tasks are backlogged
1368 > // here, so we include this queue's stats if the version to describe is a current/ramping version.
1369 > dbq := pm.defaultQueue()
1370 > if dbq == nil {
1371 return nil, errDefaultQueueNotInit
1372 }
1373 > unversionedStatsByPriority = dbq.GetStatsByPriority(true) task_queue_partition_manager.go
1374 >
1375 > userData, _, err := pm.GetUserDataManager().GetUserData()
1376 > if err != nil {
1377 return nil, err
1378 }
1379 > perType := userData.GetData().GetPerType()[int32(pm.Partition().TaskType())] task_queue_partition_manager.go
1380 > deploymentData := perType.GetDeploymentData()
1381 >
1382 > currentVersion, _, _, rampingVersion, isRamping, rampPercentage, _, _ =
1383 > worker_versioning.CalculateTaskQueueVersioningInfo(deploymentData)
1384 >
1385 > // Technically, one could have a current version of "unversioned" which shall make currentExists false according
1386 > // to the current logic. However, as of now, the user cannot query the stats of the "unversioned" version so this
1387 > // logic is fine. In other words, this logic is used to only attribute the unversioned backlog to the current version
1388 > // when current version is NOT "unversioned".
1389 > //
1390 > // When the ramping version is "unversioned", isRamping is true which shall make the attribution logic work as expected.
1391 > currentExists = currentVersion != nil
1392 > rampingExists = isRamping && rampPercentage > 0
1393 >
1394 > // Split the unversioned queue's stats per priority so TaskQueueStatsByPriorityKey can
1395 > // be adjusted consistently with TaskQueueStats.
1396 > unversionedCurrentShareByPriority = map[int32]*taskqueuepb.TaskQueueStats{}
1397 > unversionedRampingShareByPriority = map[int32]*taskqueuepb.TaskQueueStats{}
1398 > if rampingExists {
1399 unversionedCurrentShareByPriority, unversionedRampingShareByPriority =
1400 splitStatsByPriorityByRampPercentage(unversionedStatsByPriority, rampPercentage)
1401 > } else if currentExists { task_queue_partition_manager.go
1402 // If there exist no ramping version, we attribute the entire unversioned backlog to the current version.
1403 unversionedCurrentShareByPriority = cloneStatsByPriority(unversionedStatsByPriority)
1405 }
1406
1407 > versionsInfo := make(map[string]*taskqueuespb.TaskQueueVersionInfoInternal, len(versions)) task_queue_partition_manager.go
1408 > for v := range versions {
1409 > vInfo := &taskqueuespb.TaskQueueVersionInfoInternal{
1410 > PhysicalTaskQueueInfo: &taskqueuespb.PhysicalTaskQueueInfo{},
1411 > }
1412 >
1413 > // `getPhysicalQueue` always needs the right buildID passed to function correctly. If the version is a worker-deployment version and an empty buildID is passed,
1414 > // the function returns the default queue which is not what we want.
1415 > // The following assigns buildID to either a v2 based buildID or a buildID part of a worker-deployment version.
1416 > buildID := v.BuildId()
1417 > if v.Deployment() != nil {
1418 buildID = v.Deployment().BuildId
1419 }
1420
1421 > physicalQueue, err := pm.getPhysicalQueue(ctx, buildID, v.Deployment()) task_queue_partition_manager.go
1422 > if err != nil {
1423 return nil, err
1424 }
1425 > if reportPollers { task_queue_partition_manager.go
1426 vInfo.PhysicalTaskQueueInfo.Pollers = physicalQueue.GetAllPollerInfo()
1427 }
1428 > if reportStats { task_queue_partition_manager.go
1429 > physicalStatsByPriority := physicalQueue.GetStatsByPriority(true)
1430 >
1431 > // Clone the physical queue's stats by priority so we can adjust (either add, subtract) them based on the
1432 > // attribution model defined below.
1433 > adjustedStatsByPriority := cloneStatsByPriority(physicalStatsByPriority)
1434 >
1435 > // Attribution model (applied per-priority):
1436 > // - If current and/or ramping deployment versions exist, we first "give away" a portion of the
1437 > // unversioned queue's per-priority stats.
1438 > //
1439 > // Depending on the version described, we have the following options:
1440 > // - For the unversioned version itself, subtract the given-away portion (so we don't double count).
1441 > // - For current/ramping versions, add their share on top of their physical queue stats.
1442 > deploymentVersion := worker_versioning.DeploymentVersionFromDeployment(v.Deployment())
1443 >
1444 > isUnversionedDescribe := deploymentVersion == nil
1445 > isCurrentDescribe := deploymentVersion.GetDeploymentName() == currentVersion.GetDeploymentName() &&
1446 > deploymentVersion.GetBuildId() == currentVersion.GetBuildId()
1447 >
1448 > // "Ramping to unversioned" is represented by "rampingExists==true AND rampingVersion==nil".
1449 > // In that case, the ramp share should remain attributed to the unversioned queue stats and
1450 > // there is no separate versioned queue to merge that share into.
1451 > isRampingToUnversioned := rampingExists && rampingVersion == nil
1452 > isRampingDescribe := deploymentVersion.GetDeploymentName() == rampingVersion.GetDeploymentName() &&
1453 > deploymentVersion.GetBuildId() == rampingVersion.GetBuildId()
1454 >
1455 > if isUnversionedDescribe {
1456 // Reduce unversioned stats by any shares attributed to versioned queues.
1457 if currentExists {
1469 }
1470
1471 > vInfo.PhysicalTaskQueueInfo.TaskQueueStatsByPriorityKey = adjustedStatsByPriority task_queue_partition_manager.go
1472 > vInfo.PhysicalTaskQueueInfo.TaskQueueStats = aggregateStats(adjustedStatsByPriority)
1473 }
1474 > if internalTaskQueueStatus { task_queue_partition_manager.go
1475 vInfo.PhysicalTaskQueueInfo.InternalTaskQueueStatus = physicalQueue.GetInternalTaskQueueStatus()
1476 }
1480 // the full worker-deployment version string is used as an entry in the versionsInfo map. Moreover, to keep things backwards compatible, users requesting
1481 // information for non-deployment related builds will only see the buildID as an entry in the versionsInfo map.
1482 > bid := v.BuildId() task_queue_partition_manager.go
1483 > if v.Deployment() != nil {
1484 bid = worker_versioning.ExternalWorkerDeploymentVersionToString(worker_versioning.ExternalWorkerDeploymentVersionFromDeployment(v.Deployment()))
1485 }
1486 > versionsInfo[bid] = vInfo task_queue_partition_manager.go
1487 >
1488 > if !skipMarkAlive {
1489 // Skipped by periodic metrics emission to avoid resetting the idle timeout,
1490 // which would prevent queues from ever being unloaded.
1493 }
1494
1495 > return &matchingservice.DescribeTaskQueuePartitionResponse{ task_queue_partition_manager.go
1496 > VersionsInfoInternal: versionsInfo,
1497 > ScaleInfo: pm.userDataManager.PartitionScale(),
1498 > }, nil
1499 }
1500
1724 }
1725
1726 > func cloneTaskQueueStats(in *taskqueuepb.TaskQueueStats) *taskqueuepb.TaskQueueStats { task_queue_partition_manager.go
1727 > if in == nil {
1728 return &taskqueuepb.TaskQueueStats{ApproximateBacklogAge: durationpb.New(0)}
1729 }
1730 > age := in.GetApproximateBacklogAge() task_queue_partition_manager.go
1731 > if age == nil {
1732 age = durationpb.New(0)
1733 }
1734 > return &taskqueuepb.TaskQueueStats{ task_queue_partition_manager.go
1735 > ApproximateBacklogCount: in.GetApproximateBacklogCount(),
1736 > ApproximateBacklogAge: durationpb.New(age.AsDuration()),
1737 > TasksAddRate: in.GetTasksAddRate(),
1738 > TasksDispatchRate: in.GetTasksDispatchRate(),
1739 > }
1740 }
1741
1742 > func cloneStatsByPriority(in map[int32]*taskqueuepb.TaskQueueStats) map[int32]*taskqueuepb.TaskQueueStats { task_queue_partition_manager.go
1743 > out := make(map[int32]*taskqueuepb.TaskQueueStats, len(in))
1744 > for pri, s := range in {
1745 > out[pri] = cloneTaskQueueStats(s)
1746 > }
1747 > return out
1748 }
1749
go.temporal.io/server/service/matching/physical_task_queue_manager.go 15 introduced LOC · 4 ranges

Open complete file

690
691 if includeRates {
692 > c.taskTrackerLock.Lock() physical_task_queue_manager.go
693 > for pri, tt := range c.tasksAdded {
694 util.GetOrSetNew(stats, int32(pri)).TasksAddRate = tt.rate()
695 }
696 > for pri, tt := range c.tasksDispatched { physical_task_queue_manager.go
697 util.GetOrSetNew(stats, int32(pri)).TasksDispatchRate = tt.rate()
698 }
699 > rateLimitingActive := c.tasksRateLimited.rate() > 0 physical_task_queue_manager.go
700 > c.taskTrackerLock.Unlock()
701 >
702 > for _, s := range stats {
703 > s.RateLimitingActive = rateLimitingActive
704 > }
705 }
706
993 }
994
995 > func aggregateStats(stats map[int32]*taskqueuepb.TaskQueueStats) *taskqueuepb.TaskQueueStats { physical_task_queue_manager.go
996 > result := &taskqueuepb.TaskQueueStats{ApproximateBacklogAge: durationpb.New(0)}
997 > for _, s := range stats {
998 > taskqueue.MergeStats(result, s)
999 > }
1000 > return result
1001 }

Direct hierarchy

Parents

Children