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