Message distribution algorithm from multiple queues considering quotas and full channel load

Given:

1. A channel for sending useful messages, with a fixed bandwidth n – the maximum number of messages that can be transmitted in one cycle.
2. An arbitrary number of queues m. Messages in the queue appear asynchronously and in different volumes; new queues can appear at any moment.
3. Each queue has a unique identifier (an integer).
4. At the time of the request, the current size of each queue is known – the number of waiting messages.
5. Each queue has a positive quota (weight), reflecting the minimum share of the channel that the queue must receive.

Task: develop an algorithm that, at each iteration, determines how many messages should be taken from each queue and sent to the channel, adhering to three requirements:

1. Work‑conserving – use the channel as efficiently as possible. If the total number of messages in all queues exceeds n, then exactly n messages should be sent to the channel.
2. Starvation‑free – ensure that queues with small shares also receive messages. For example, if one queue has a share of 1000, and another – 0.00001, messages from the second queue should still be sent.
3. Fairness – ensure that each queue gets access to the channel in accordance with its quota.

Example operation with channel bandwidth 10:

• One queue, quota 0.5, 100 messages → 10 messages are sent (quota does not limit, channel is fully loaded).
• Two queues, both with quota 0.5, 100 messages each → 5 messages from each.
• Two queues, quotas 0.2 and 0.8, 100 messages each → 2 messages from the first, 8 from the second.
• Two queues, quotas 0.25 and 1, 100 messages each → 2 messages from the first, 8 from the second.
• Ten queues, each with quota 1 and 100 messages → one message is taken from each of every ten queues, then shifted by one, and so on.

// Main message dispatch loop, which calculates the required number of messages for each queue at each iteration.
function dispatcher(Source $source) {
    $messages = [];
    while (true) {
        // Simulate message arrival in queues
        $source->next();
        // Calculate the number of messages to extract from each queue
        $batchSizes = calculateMessageBatchSizes($source->queueSizes(), $source->queueQuotas());
        $source->printStats($batchSizes);
        // Extract the required number of messages from each queue and add them to the overall array
        foreach ($batchSizes as $queueId => $batchSize) {
            if ($batchSize <= 0) {
                continue;
            }
            $messages = array_merge(
                $messages,
                $source->extractMessagesFromQueue($queueId, $batchSize)
            );
        }
        // Send messages if any
        if ($messages) {
            sendMessages($messages);
            $messages = [];
        }
    }
}

/**
 * @param array<int, int> $queueSizes
 * @param array<int, float> $queueQuotas
 * @param int $bandwidth
 * @return array
 */
function calculateMessageBatchSizes(array $queueSizes, array $queueQuotas, int $bandwidth = 20): array
{