Method EnqueueRequest

pkg/scheduler/queue/queue.go:91–123

EnqueueRequest puts the request into the queue. MaxQueries is user-specific value that specifies how many queriers can this user use (zero or negative = all queriers). It is passed to each EnqueueRequest, because it can change between calls. If request is successfully enqueued, successFn is called

(userID string, req Request, maxQueriers float64, successFn func())

//
// If request is successfully enqueued, successFn is called with the lock held, before any querier can receive the request.
func (q *RequestQueue) EnqueueRequest(userID string, req Request, maxQueriers float64, successFn func()) error {
	q.mtx.Lock()
	defer q.mtx.Unlock()

	if q.stopped {
		return ErrStopped
	}

	maxQuerierCount := util.DynamicShardSize(maxQueriers, len(q.queues.queriers))
	queue := q.queues.getOrAddQueue(userID, maxQuerierCount)
	maxOutstandingRequests := q.queues.limits.MaxOutstandingPerTenant(userID)
	priority := strconv.FormatInt(req.Priority(), 10)

	if queue == nil {
		// This can only happen if userID is "".
		return errors.New("no queue found")
	}

	q.totalRequests.WithLabelValues(userID, priority).Inc()

	if queue.length() >= maxOutstandingRequests {
		q.discardedRequests.WithLabelValues(userID, priority).Inc()
		return ErrTooManyRequests
	}

	queue.enqueueRequest(req)
	q.cond.Broadcast()
	// Call this function while holding a lock. This guarantees that no querier can fetch the request before function returns.
	if successFn != nil {
		successFn()
	}
	return nil
}

// 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.