1520
return ctx.Err()
1521
1523
>
prevBacklogPriority = pm.updateEphemeralDataIteration(prevBacklogPriority)
1524
}
1525
}
1526
}
1527
1528
>
func (pm *taskQueuePartitionManagerImpl) updateEphemeralDataIteration(prevBacklogPriority map[PhysicalTaskQueueVersion]int64) map[PhysicalTaskQueueVersion]int64 {
task_queue_partition_manager.go
1529
>
negligibleAge := pm.config.BacklogNegligibleAge()
1530
>
backlogPriority := make(map[PhysicalTaskQueueVersion]int64)
1531
>
1532
>
setLevels := func(vk PhysicalTaskQueueVersion, vq physicalTaskQueueManager) {
1533
>
var levels int64
1534
>
for key, stats := range vq.GetStatsByPriority(false) {
1535
>
if key < 64 && stats.ApproximateBacklogAge.AsDuration() > negligibleAge {
1536
levels = levels | 1<<key
1537
}
1538
}
1540
backlogPriority[vk] = levels
1541
}
1542
}
1543
1545
>
if dbq == nil {
1546
return prevBacklogPriority // shouldn't happen, we only start after initialization
1547
}
1549
>
1550
>
pm.versionedQueuesLock.RLock()
1551
>
for vk, vq := range pm.versionedQueues {
1552
if vk.buildId == "" || vk.deploymentSeriesName == "" {
1553
continue // v3 only