ClickHouse Shares a CPU Budget Between Queries with Fair CPU Slots

// src/Common/ISlotControl.h:14
// Interfaces for abstract "slot" allocation and control.
// Slot is a virtual entity existing in a limited amount (CPUs or memory chunks, etc).
//
// Every slot can be in one of the following states:
//  * free: slot is available to be allocated.
//  * allocated: slot is allocated to a specific ISlotAllocation.
//
// Allocated slots can be in one of the following states:
//  * granted: allocated, but not yet acquired.
//  * acquired: a granted slot becomes acquired (e.g. by using IAcquiredSlot).
//
// Example for CPU (see ConcurrencyControl.h). Every slot represents one CPU in the system.
// Slot allocation is a request to allocate specific number of CPUs for a specific query.
// Acquired slot is an entity that is held by a thread as long as it is running. This allows
// total number of threads in the system to be limited and the distribution process to be controlled.
//
// TODO:
// - for memory allocations - variable size of slots (in bytes).

// src/Common/ISlotControl.h:39
/// Acquired slot holder. Slot is considered to be acquired as long as the object exists.
class IAcquiredSlot : public std::enable_shared_from_this<IAcquiredSlot>, boost::noncopyable
{
public:
    const size_t slot_id; /// Unique identifier of a slot within specific ISlotAllocation.

    explicit IAcquiredSlot(size_t slot_id_)
        : slot_id(slot_id_)
    {}

    virtual ~IAcquiredSlot() = default;
};

// src/Common/ISlotControl.h:87
/// Request for allocation of slots from ISlotControl.
/// Allows for more slots to be acquired and the whole request to be canceled.
class ISlotAllocation : public std::enable_shared_from_this<ISlotAllocation>, boost::noncopyable
{
public:
    virtual ~ISlotAllocation() = default;

    /// Free the allocated slots, cancel slot requests and wake up preempted threads.
    virtual void free() {}

    /// Take one already granted slot if available.
    [[nodiscard]] virtual AcquiredSlotPtr tryAcquire() = 0;

    /// Take one granted slot or wait until it is available.
    [[nodiscard]] virtual AcquiredSlotPtr acquire() = 0;

    /// Adjust the maximum number of slots the allocation wants.
    ///
    /// The allocation's `max` (set initially by allocate(min, max)) becomes mutable: the
    /// consumer can raise or lower it to express "I want up to N slots" at runtime.
    ///
    /// Contract for implementations:
    ///  - `new_max` MUST be > 0. Implementations may chassert this. A request for zero slots
    ///    has no sensible behavior here (use allocation destruction to release the work).
    ///  - Growing `max`: if the allocation was previously saturated (had received its full
    ///    max and was no longer a waiter), it MUST be re-added to the scheduler's waiter
    ///    list so it may receive further grants.
    ///  - Shrinking `max`: does NOT reclaim already-granted slots; the scheduler simply
    ///    stops issuing new grants beyond `new_max`. Already-acquired slots remain valid.
    ///  - Idempotent: setMax(current) is a no-op.
    ///  - Must NOT be called while holding the implementation's internal scheduler lock —
    ///    implementations may take that lock internally.
    /// Default implementation is a no-op (for allocations that don't need dynamic sizing).
    virtual void setMax(SlotCount /*new_max*/) {}

    /// For tests. Returns true iff resource request is currently enqueued into the scheduler.
    virtual bool isRequesting() const { return false; }
};

// src/Common/ConcurrencyControl.h:63
struct ConcurrencyControlState
{
    ConcurrencyControlState();

    SlotCount available(std::unique_lock<std::mutex> &) const;

    mutable std::mutex mutex;

    // Slot counting
    SlotCount max_concurrency = UnlimitedSlots;
    SlotCount cur_concurrency = 0;
    CurrentMetrics::Increment max_concurrency_metric;

    // Number of live allocations (of all schedulers), used to estimate the fair share of a new allocation
    size_t allocations = 0;

    // Rollback lever for the lazy allocation strategy. When false, pipelines bypass the
    // lazy `allocate(1, 1) + setMax` growth path and use eager `allocate(1, num_threads)`
    // directly. Read without locking (relaxed).
    std::atomic<bool> lazy_allocation{true};
};

// src/Core/ServerSettings.cpp:1077
    DECLARE(UInt64, concurrent_threads_soft_limit_num, 0, R"(
The maximum number of query processing threads, excluding threads for retrieving data from remote servers, allowed to run all queries. This is not a hard limit. In case if the limit is reached the query will still get at least one thread to run. Query can upscale to desired number of threads during execution if more threads become available.

<Note>
A value of `0` (default) means unlimited.
</Note>
)", 0) \
    DECLARE(UInt64, concurrent_threads_soft_limit_ratio_to_cores, 2, "Same as [`concurrent_threads_soft_limit_num`](#concurrent_threads_soft_limit_num), but with ratio to cores.", 0) \

// src/Interpreters/Context.cpp:2981
    // Set the limit
    SlotCount concurrent_threads_soft_limit = UnlimitedSlots;
    if (num > 0 && num < concurrent_threads_soft_limit)
        concurrent_threads_soft_limit = num;
    if (ratio_to_cores > 0)
    {
        auto value = ratio_to_cores * getNumberOfCPUCoresToUse();
        if (value > 0 && value < concurrent_threads_soft_limit)
            concurrent_threads_soft_limit = value;
    }
    ConcurrencyControl::instance().setMaxConcurrency(concurrent_threads_soft_limit);

// src/Common/ConcurrencyControl.cpp:784
void ConcurrencyControl::setMaxConcurrency(SlotCount value)
{
    std::unique_lock lock{state.mutex};
    state.max_concurrency = std::max<SlotCount>(1, value); // never allow max_concurrency to be zero
    state.max_concurrency_metric.changeTo(state.max_concurrency == UnlimitedSlots ? 0 : state.max_concurrency);
    schedule(lock);
}

// src/Common/ConcurrencyControl.h:46
 * There is a scheduler of CPU slots. It could be set with `setScheduler(name)`.
 * Possible values:
 *  - "round_robin":
 *    Released slots are distributed between waiting allocations in a round-robin manner to provide fairness.
 *    Oversubscription is possible: total amount of allocated slots can exceed `setMaxConcurrency(limit)`
 *    because `min` amount of slots is allocated for each query unconditionally.
 *  - "fair_round_robin":
 *    Also uses round-robin, but `min` slot are NOT holding real CPU slots.
 *    This way all `min` slots do not count into overall number of allocated slots. This leads to more fair competition.
 *    There is no oversubscription: total amount of allocated slots CANNOT exceed `setMaxConcurrency(limit)`.
 *  - "max_min_fair":
 *    Similar to "fair_round_robin", but released slots are always granted to the allocation with the minimum
 *    number of currently allocated slots. This provides better fairness under high oversubscription scenarios
 *    where many queries compete for limited CPU slots.
 */

// src/Core/ServerSettings.cpp:1092
    DECLARE(String, concurrent_threads_scheduler, "max_min_fair", R"(
The policy on how to perform a scheduling of CPU slots specified by `concurrent_threads_soft_limit_num` and `concurrent_threads_soft_limit_ratio_to_cores`. Algorithm used to govern how limited number of CPU slots are distributed among concurrent queries. Scheduler may be changed at runtime without server restart.

Possible values:

- `round_robin` — Every query with setting `use_concurrency_control` = 1 allocates up to `max_threads` CPU slots. One slot per thread. On contention CPU slot are granted to queries using round-robin. Note that the first slot is granted unconditionally, which could lead to unfairness and increased latency of queries having high `max_threads` in presence of high number of queries with `max_threads` = 1.
- `fair_round_robin` — Every query with setting `use_concurrency_control` = 1 allocates up to `max_threads - 1` CPU slots. Variation of `round_robin` that does not require a CPU slot for the first thread of every query. This way queries having `max_threads` = 1 do not require any slot and could not unfairly allocate all slots. There are no slots granted unconditionally.
- `max_min_fair` — Every query with setting `use_concurrency_control` = 1 allocates up to `max_threads - 1` CPU slots. Similar to `fair_round_robin`, but released slots are always granted to the query with the minimum number of currently allocated slots. This provides better fairness under high oversubscription, where many queries compete for limited CPU slots. Short-running queries are not penalized by long-running queries that have accumulated more slots over time.

// src/Common/ConcurrencyControl.cpp:160
SlotAllocationPtr ConcurrencyControlRoundRobinScheduler::allocate(std::unique_lock<std::mutex> & lock, SlotCount min, SlotCount max)
{
    // Grant exactly `min` slots up front. Oversubscription is allowed (RR doesn't gate min
    // by capacity). Then run `schedule` so that any remaining capacity flows into this
    // allocation up to its `max` -- this is what gives callers the eager `allocate(min, max)`
    // semantics they expect when they don't intend to grow via `setMax`.
    const SlotCount granted = min;
    state.cur_concurrency += granted;
    ProfileEvents::increment(ProfileEvents::ConcurrencyControlSlotsGranted, min);

    SlotAllocationPtr alloc;
    if (granted < max)
    {
        alloc = SlotAllocationPtr(new Allocation(*this, max, granted,
            waiters.insert(cur_waiter, nullptr /* pointer is set by Allocation ctor */)));
    }
    else
    {
        alloc = SlotAllocationPtr(new Allocation(*this, max, granted));
    }
    // Emit delayed metrics for any portion of `max - min` that current capacity can't
    // cover. This catches the eager rollback path (`allocate(1, num_threads)`) which
    // never calls `setMax` and would otherwise leave the metric at zero under pressure.
    if (max > min)
    {
        const SlotCount delta = max - min;
        const SlotCount available = state.available(lock);
        emitDelayedOnGrow(static_cast<Allocation *>(alloc.get())->query_counted, delta, available);
    }
    // Fill the new allocation up to `max` from available capacity (also benefits any other
    // waiters that have free room). For lazy callers (allocate(1, 1)), this is a no-op.
    parent.schedule(lock);
    return alloc;
}

// src/Common/ConcurrencyControl.cpp:214
// Round-robin scheduling of available slots among waiting allocations.
// Single mode: walk the waiter list, grant up to each waiter's current `limit` while
// capacity allows. The limit IS the demand signal (set by the caller via setMax).
void ConcurrencyControlRoundRobinScheduler::schedule(std::unique_lock<std::mutex> &)
{
    while (!waiters.empty() && state.cur_concurrency < state.max_concurrency)
    {
        state.cur_concurrency++;
        if (cur_waiter == waiters.end())
            cur_waiter = waiters.begin();
        Allocation * allocation = *cur_waiter;
        bool still_waiter = allocation->grant();
        if (still_waiter)
        {
            ++cur_waiter;
        }
        else
        {
            cur_waiter = waiters.erase(cur_waiter);
        }
    }
}

// src/Common/ConcurrencyControl.cpp:406
SlotAllocationPtr ConcurrencyControlFairRoundRobinScheduler::allocate(std::unique_lock<std::mutex> & lock, SlotCount min, SlotCount max)
{
    // Min slots are non-competing -- they don't count toward the global limit. Grant zero
    // competing slots up front; `schedule` below then fills up to the competing limit.
    SlotCount limit = max - min;
    SlotCount granted = 0;
    ProfileEvents::increment(ProfileEvents::ConcurrencyControlSlotsGranted, min);

    SlotAllocationPtr alloc;
    if (granted < limit)
    {
        alloc = SlotAllocationPtr(new Allocation(*this, min, max, granted,
            waiters.insert(cur_waiter, nullptr /* pointer is set by Allocation ctor */)));
    }
    else
    {
        alloc = SlotAllocationPtr(new Allocation(*this, min, max, granted));
    }
    // See RR allocate for the rationale. FRR's competing demand is `max - min` (= limit).
    if (limit > 0)
    {
        const SlotCount available = state.available(lock);
        emitDelayedOnGrow(static_cast<Allocation *>(alloc.get())->query_counted, limit, available);
    }
    parent.schedule(lock);
    return alloc;
}

// src/Common/ConcurrencyControl.h:371
    // Use boost intrusive set sorted by (allocated, sequence_number) to efficiently find minimum allocation
    // sequence_number ensures FIFO order when allocated counts are equal
    struct AllocationCompare
    {
        bool operator()(const Allocation & lhs, const Allocation & rhs) const;
    };
    using Waiters = boost::intrusive::set<
        Allocation,
        boost::intrusive::compare<AllocationCompare>,
        boost::intrusive::member_hook<Allocation, boost::intrusive::set_member_hook<>, &Allocation::waiters_hook>
    >;

// src/Common/ConcurrencyControl.cpp:679
// Max-min fair: walk waiters in (allocated, sequence) order and grant up to each one's
// current `limit` while capacity allows. The sort key changes on grant (`allocated++`), so
// we erase before `grant` and re-insert if still a waiter.
void ConcurrencyControlMaxMinFairScheduler::schedule(std::unique_lock<std::mutex> &)
{
    while (!waiters.empty() && state.cur_concurrency < state.max_concurrency)
    {
        state.cur_concurrency++;

        auto it = waiters.begin();
        Allocation & allocation = *it;

        removeWaiterLocked(&allocation);
        chassert(!allocation.waiters_hook.is_linked());

        bool still_waiter = allocation.grant();
        if (still_waiter)
            addWaiterLocked(&allocation);
    }
}

// src/Common/ConcurrencyControl.cpp:237
// Raise or lower the allocation's max slot ceiling. Emits SlotsDelayed/QueriesDelayed when
// the grow request exceeds current capacity. Grow past saturation re-inserts into waiters;
// shrink past saturation removes from waiters. Already-granted slots are not reclaimed.
//
// MUST NOT be called while holding state.mutex -- we take it here internally.
void ConcurrencyControlRoundRobinScheduler::Allocation::setMax(SlotCount new_max)
{
    chassert(new_max > 0);
    std::unique_lock lock{parent.state.mutex};
    bool need_schedule = false;
    {
        std::unique_lock alock{mutex};
        if (new_max == limit)
            return;
        bool was_waiter = (allocated < limit);
        bool will_be_waiter = (allocated < new_max);
        SlotCount old_limit = limit;
        limit = new_max;
        if (!was_waiter && will_be_waiter)
        {
            // Grew past saturation -- re-insert.
            waiter = parent.waiters.insert(parent.cur_waiter, this);
            need_schedule = true;
        }
        else if (was_waiter && !will_be_waiter)
        {
            // Shrunk past saturation -- remove from waiters so `free` sees consistent state.
            if (parent.cur_waiter == waiter)
                parent.cur_waiter = parent.waiters.erase(waiter);
            else
                parent.waiters.erase(waiter);
        }
        else if (was_waiter && will_be_waiter && new_max > old_limit)
        {
            // Still a waiter but grew.
            need_schedule = true;
        }

        if (new_max > old_limit)
        {
            // Only count slots beyond what was previously asked-for AND beyond what's
            // already allocated. `allocated > old_limit` is possible after a prior
            // shrink (setMax does not reclaim slots); without the std::max, the
            // subtraction would underflow on this path.
            const SlotCount baseline = std::max(old_limit, allocated);
            if (new_max > baseline)
            {
                const SlotCount delta = new_max - baseline;
                const SlotCount available = parent.state.available(lock);
                emitDelayedOnGrow(query_counted, delta, available);
            }
        }
    }
    if (need_schedule)
        parent.parent.schedule(lock);
}

// src/Common/ConcurrencyControl.cpp:792
SlotCount ConcurrencyControl::getFairShare() const
{
    std::unique_lock lock{state.mutex};
    if (state.max_concurrency == UnlimitedSlots)
        return UnlimitedSlots;
    return std::max<SlotCount>(1, state.max_concurrency / (state.allocations + 1));
}

// src/Common/ISlotControl.h:138
/// Allocation that grants all the slots immediately on creation
class GrantedAllocation : public ISlotAllocation
{
public:
    explicit GrantedAllocation(SlotCount granted_)
        : total(granted_)
        , granted(granted_)
    {}

    [[nodiscard]] AcquiredSlotPtr tryAcquire() override
    {
        SlotCount value = granted.load();
        while (value)
        {
            if (granted.compare_exchange_strong(value, value - 1))
                return std::make_shared<IAcquiredSlot>(total - value);
        }
        return {};
    }

    [[nodiscard]] AcquiredSlotPtr acquire() override
    {
        auto result = tryAcquire();
        chassert(result);
        return result;
    }

private:
    const SlotCount total; // thread-safe constant total number of slots
    std::atomic<SlotCount> granted; // allocated, but not yet acquired
};

// src/Common/ConcurrencyControl.cpp:842
void ConcurrencyControl::schedule(std::unique_lock<std::mutex> & lock)
{
    // Run all three schedulers each round so transitions (setScheduler) leave behind no
    // stranded waiters in the previous mode.
    switch (scheduler)
    {
        case Scheduler::RoundRobin:
            fair_round_robin.schedule(lock);
            max_min_fair.schedule(lock);
            round_robin.schedule(lock);
            return;

        +-------------------- ConcurrencyControl (one mutex) --------------------+
        |  max_concurrency = 2 * ncores (default)                             |
        |  cur_concurrency = competing slots currently allocated              |
        |  allocations = live queries                                          |
        |                                                                        |
        |  max_min_fair waiters: intrusive set sorted by (allocated, seq)      |
        |      [ Q3: 2 slots ] <-> [ Q7: 0 slots ] <-> [ Q9: 1 slot ]          |
        +------------------------------------------------------------------------+
             ^            ^                 |
             |            |                 |  setMax(n) grows the ceiling
        allocate(1,1)  tryAcquire()             |
             |            |                 v
   +---------+------+     |        +------------------------+
   | Query A       |     |        | Query B                 |
   | master: 1 slot|<----+        | master: 1 slot          |
   | ceiling: 1->8  |  released  | ceiling: 1->8           |
   +----------------+            +-------------------------+
                                        ^
                                        |
                        a released slot goes to the allocation
                        with the FEWEST slots (max_min_fair)

SELECT event, value
FROM system.events
WHERE event ILIKE 'ConcurrencyControl%'
ORDER BY event;

-- event                                  -- value
-- ConcurrencyControlSlotsGranted         -- 91        (slots handed out)
-- ConcurrencyControlSlotsDelayed         -- 12        (slots wanted but not available)
-- ConcurrencyControlSlotsAcquired        -- 88
-- ConcurrencyControlSlotsAcquiredNonCompeting -- 7   (the master-thread slots)
-- ConcurrencyControlQueriesDelayed       -- 3         (queries that hit the wall)
-- ConcurrencyControlWaitMicroseconds     -- 51120
-- ConcurrencyControlParks                -- 402       (CPU leases parked for non-CPU waits)
-- ConcurrencyControlUnparks              -- 401

// src/Common/ProfileEvents.cpp:1664
    M(ConcurrencyControlQueriesDelayed, "Number of queries delayed by CPU slot capacity pressure. Contention indicator — excludes lazy-upscale waits on an uncontested server.", ValueType::Number) \

// src/Common/ConcurrencyControl.cpp:44
namespace
{
    /// Record an attempted ceiling raise that available capacity couldn't fully satisfy.
    /// Bumps `ConcurrencyControlSlotsDelayed` by the shortfall and, the first time it
    /// observes pressure on a given allocation, bumps `ConcurrencyControlQueriesDelayed`
    /// exactly once. `query_counted` lives on the allocation so the once-per-allocation
    /// guarantee survives multiple calls from `allocate` and subsequent `setMax` grows.
    void emitDelayedOnGrow(bool & query_counted, SlotCount wanted, SlotCount available)
    {
        if (wanted <= available)
            return;
        ProfileEvents::increment(ProfileEvents::ConcurrencyControlSlotsDelayed, wanted - available);
        if (!query_counted)
        {
            ProfileEvents::increment(ProfileEvents::ConcurrencyControlQueriesDelayed);
            query_counted = true;
        }
    }
}

SELECT name, value, changed FROM system.server_settings
WHERE name IN ('concurrent_threads_soft_limit_num',
               'concurrent_threads_soft_limit_ratio_to_cores',
               'concurrent_threads_lazy_allocation',
               'concurrent_threads_scheduler');