277
}
278
279
>
func (c *priBacklogManagerImpl) BacklogStatsByPriority() map[int32]*taskqueuepb.TaskQueueStats {
pri_backlog_manager.go
280
>
c.subqueueLock.Lock()
281
>
defer c.subqueueLock.Unlock()
282
>
283
>
result := make(map[int32]*taskqueuepb.TaskQueueStats)
284
>
backlogCounts := c.db.getApproximateBacklogCountsBySubqueue()
285
>
for subqueueIdx, priorityKey := range c.priorityBySubqueue {
286
>
pk := int32(priorityKey)
287
>
288
>
// Note that there could be more than one subqueue for the same priority.
289
>
if _, ok := result[pk]; !ok {
290
>
result[pk] = &taskqueuepb.TaskQueueStats{
291
>
// TODO(pri): returning 0 to match existing behavior, but maybe emptyBacklogAge would
292
>
// be more appropriate in the future.
293
>
ApproximateBacklogAge: durationpb.New(0),
294
>
}
295
>
}
296
297
// Add backlog counts together across all subqueues for the same priority.
299
>
300
>
// Find greatest backlog age for across all subqueues for the same priority.
301
>
oldestBacklogTime := c.subqueues[subqueueIdx].getOldestBacklogTime()
302
>
if !oldestBacklogTime.IsZero() {
303
oldestBacklogAge := time.Since(oldestBacklogTime)
304
if oldestBacklogAge > result[pk].ApproximateBacklogAge.AsDuration() {