Method GetNextRequestForQuerier

pkg/scheduler/queue/queue.go:128–181

GetNextRequestForQuerier find next user queue and takes the next request off of it. Will block if there are no requests. By passing user index from previous call of this method, querier guarantees that it iterates over all users fairly. If querier finds that request from the user is already expired,

(ctx context.Context, last UserIndex, querierID string)

// By passing user index from previous call of this method, querier guarantees that it iterates over all users fairly.
// If querier finds that request from the user is already expired, it can get a request for the same user by using UserIndex.ReuseLastUser.
func (q *RequestQueue) GetNextRequestForQuerier(ctx context.Context, last UserIndex, querierID string) (Request, UserIndex, error) {
	q.mtx.Lock()
	defer q.mtx.Unlock()

	querierWait := false

FindQueue:
	// We need to wait if there are no users, or no pending requests for given querier.
	for (q.queues.len() == 0 || querierWait) && ctx.Err() == nil && !q.stopped {
		querierWait = false
		q.cond.Wait()
	}

	if q.stopped {
		return nil, last, ErrStopped
	}

	if err := ctx.Err(); err != nil {
		return nil, last, err
	}

	for {
		queue, userID, idx := q.queues.getNextQueueForQuerier(last.last, querierID)
		last.last = idx
		if queue == nil {
			break
		}

		// Pick next request from the queue.
		for {
			minPriority, matchMinPriority := q.getPriorityForQuerier(userID, querierID)
			request := queue.dequeueRequest(minPriority, matchMinPriority)
			if request == nil {
				// The queue does not contain request with the priority, wait for more requests
				querierWait = true
				goto FindQueue
			}

			if queue.length() == 0 {
				q.queues.deleteQueue(userID)
			}

			// Tell close() we've processed a request.
			q.cond.Broadcast()

			return request, last, nil
		}
	}

	// There are no unexpired requests, so we can get back
	// and wait for more requests.
	querierWait = true
	goto FindQueue
}

func (q *RequestQueue) getPriorityForQuerier(userID string, querierID string) (int64, bool) {
	if priority, ok := q.queues.userQueues[userID].reservedQueriers[querierID]; ok {