Method rebuildPendingIndexes

modules/backendscheduler/work/work.go:602–640

rebuildPendingIndexes rebuilds the pendingBlocks, pendingByTenant, and runningBlocks indexes from all shards' Pending and Jobs maps. Caller must hold w.mtx (e.g. during LoadFromLocal).

()

// rebuildPendingIndexes rebuilds the pendingBlocks, pendingByTenant, and runningBlocks indexes
// from all shards' Pending and Jobs maps. Caller must hold w.mtx (e.g. during LoadFromLocal).
func (w *Work) rebuildPendingIndexes() {
	w.pendingMtx.Lock()
	defer w.pendingMtx.Unlock()

	w.pendingBlocks = make(map[string]string)
	w.pendingByTenant = make(map[string]map[tempopb.JobType][]string)

	for i := range ShardCount {
		shard := w.Shards[i]
		shard.mtx.Lock()
		for _, j := range shard.Pending {
			if key := j.PendingBlockKey(); key != "" {
				w.pendingBlocks[key] = j.ID
			}
			tenant := j.JobDetail.Tenant
			if w.pendingByTenant[tenant] == nil {
				w.pendingByTenant[tenant] = make(map[tempopb.JobType][]string)
			}
			w.pendingByTenant[tenant][j.Type] = append(w.pendingByTenant[tenant][j.Type], j.ID)
		}
		shard.mtx.Unlock()
	}

	w.runningBlocks = make(map[string]*Job)
	for i := range ShardCount {
		shard := w.Shards[i]
		shard.mtx.Lock()
		for _, j := range shard.Jobs {
			switch j.GetStatus() {
			case tempopb.JobStatus_JOB_STATUS_UNSPECIFIED,
				tempopb.JobStatus_JOB_STATUS_RUNNING:
				for _, key := range runningBlockKeys(j) {
					w.runningBlocks[key] = j
				}
			}
		}
		shard.mtx.Unlock()
	}
}

// AddPendingJobs adds jobs to the appropriate shards' Pending maps and updates the blocks-pending index.
func (w *Work) AddPendingJobs(jobs []*Job) error {