Qu'est-ce qu'un bon algorithme de limitation de débit?

Je pourrais utiliser un pseudo-code, ou mieux, Python. J'essaie d'implémenter une file d'attente de limitation de débit pour un bot Python IRC, et cela fonctionne partiellement, mais si quelqu'un déclenche moins de messages que la limite (par exemple, la limite de débit est de 5 messages par 8 secondes et la personne ne déclenche que 4), et le prochain déclencheur dépasse les 8 secondes (par exemple, 16 secondes plus tard), le bot envoie le message, mais la file d'attente devient pleine et le bot attend 8 secondes, même si ce n'est pas nécessaire puisque la période de 8 secondes s'est écoulée.

Voici l' algorithme le plus simple , si vous souhaitez simplement supprimer les messages lorsqu'ils arrivent trop rapidement (au lieu de les mettre en file d'attente, ce qui est logique car la file d'attente peut devenir arbitrairement grande):

rate = 5.0; // unit: messages
per  = 8.0; // unit: seconds
allowance = rate; // unit: messages
last_check = now(); // floating-point, e.g. usec accuracy. Unit: seconds

when (message_received):
  current = now();
  time_passed = current - last_check;
  last_check = current;
  allowance += time_passed * (rate / per);
  if (allowance > rate):
    allowance = rate; // throttle
  if (allowance < 1.0):
    discard_message();
  else:
    forward_message();
    allowance -= 1.0;

Il n'y a pas de structures de données, de temporisateurs, etc. dans cette solution et cela fonctionne proprement :) Pour voir cela, «allocation» croît à une vitesse de 5/8 unités par seconde au plus, soit au plus cinq unités par huit secondes. Chaque message transféré déduit une unité, vous ne pouvez donc pas envoyer plus de cinq messages toutes les huit secondes.

Notez que cela ratedevrait être un entier, c'est-à-dire sans partie décimale non nulle, sinon l'algorithme ne fonctionnera pas correctement (le taux réel ne le sera pas rate/per). Par exemple, rate=0.5; per=1.0;ne fonctionne pas car allowanceil ne passera jamais à 1.0. Mais rate=1.0; per=2.0;fonctionne très bien.

Utilisez ce décorateur @RateLimited (ratepersec) avant votre fonction qui met en file d'attente.

Fondamentalement, cela vérifie si 1 / rate sec s'est écoulé depuis la dernière fois et sinon, attend le reste du temps, sinon il n'attend pas. Cela vous limite effectivement au taux / sec. Le décorateur peut être appliqué à n'importe quelle fonction que vous souhaitez limiter au taux.

Dans votre cas, si vous voulez un maximum de 5 messages toutes les 8 secondes, utilisez @RateLimited (0,625) avant votre fonction sendToQueue.

import time

def RateLimited(maxPerSecond):
    minInterval = 1.0 / float(maxPerSecond)
    def decorate(func):
        lastTimeCalled = [0.0]
        def rateLimitedFunction(*args,**kargs):
            elapsed = time.clock() - lastTimeCalled[0]
            leftToWait = minInterval - elapsed
            if leftToWait>0:
                time.sleep(leftToWait)
            ret = func(*args,**kargs)
            lastTimeCalled[0] = time.clock()
            return ret
        return rateLimitedFunction
    return decorate

@RateLimited(2)  # 2 per second at most
def PrintNumber(num):
    print num

if __name__ == "__main__":
    print "This should print 1,2,3... at about 2 per second."
    for i in range(1,100):
        PrintNumber(i)

Un Token Bucket est assez simple à implémenter.

Commencez avec un seau avec 5 jetons.

Toutes les 5/8 secondes: si le compartiment contient moins de 5 jetons, ajoutez-en un.

Chaque fois que vous voulez envoyer un message: Si le seau a ≥1 jeton, retirez un jeton et envoyez le message. Sinon, attendez / laissez tomber le message / peu importe.

(évidemment, dans le code réel, vous utiliseriez un compteur entier au lieu de vrais jetons et vous pouvez optimiser le pas toutes les 5 / 8s en stockant les horodatages)

En relisant la question, si la limite de débit est entièrement réinitialisée toutes les 8 secondes, alors voici une modification:

Commencez par un horodatage,, last_sendà un moment il y a longtemps (par exemple, à l'époque). Commencez également avec le même seau à 5 jetons.

Frappez la règle toutes les 5/8 secondes.

Chaque fois que vous envoyez un message: Vérifiez d'abord s'il y a last_send≥ 8 secondes. Si tel est le cas, remplissez le seau (définissez-le sur 5 jetons). Deuxièmement, s'il y a des jetons dans le compartiment, envoyez le message (sinon, déposez / attendez / etc.). Troisièmement, réglé last_sendmaintenant.

Cela devrait fonctionner pour ce scénario.

J'ai en fait écrit un bot IRC en utilisant une stratégie comme celle-ci (la première approche). C'est en Perl, pas en Python, mais voici du code pour illustrer:

La première partie ici gère l'ajout de jetons au bucket. Vous pouvez voir l'optimisation de l'ajout de jetons en fonction du temps (de la 2e à la dernière ligne), puis la dernière ligne limite le contenu du seau au maximum (MESSAGE_BURST)

    my $start_time = time;
    ...
    # Bucket handling
    my $bucket = $conn->{fujiko_limit_bucket};
    my $lasttx = $conn->{fujiko_limit_lasttx};
    $bucket += ($start_time-$lasttx)/MESSAGE_INTERVAL;
    ($bucket <= MESSAGE_BURST) or $bucket = MESSAGE_BURST;

$ conn est une structure de données qui est transmise. C'est à l'intérieur d'une méthode qui s'exécute régulièrement (elle calcule quand la prochaine fois qu'elle aura quelque chose à faire et dort aussi longtemps ou jusqu'à ce qu'elle reçoive du trafic réseau). La partie suivante de la méthode gère l'envoi. C'est plutôt compliqué, car les messages sont associés à des priorités.

    # Queue handling. Start with the ultimate queue.
    my $queues = $conn->{fujiko_queues};
    foreach my $entry (@{$queues->[PRIORITY_ULTIMATE]}) {
            # Ultimate is special. We run ultimate no matter what. Even if
            # it sends the bucket negative.
            --$bucket;
            $entry->{code}(@{$entry->{args}});
    }
    $queues->[PRIORITY_ULTIMATE] = [];

C'est la première file d'attente, qui est exécutée quoi qu'il arrive. Même si cela fait tuer notre connexion pour une inondation. Utilisé pour des choses extrêmement importantes, comme répondre au PING du serveur. Ensuite, le reste des files d'attente:

    # Continue to the other queues, in order of priority.
    QRUN: for (my $pri = PRIORITY_HIGH; $pri >= PRIORITY_JUNK; --$pri) {
            my $queue = $queues->[$pri];
            while (scalar(@$queue)) {
                    if ($bucket < 1) {
                            # continue later.
                            $need_more_time = 1;
                            last QRUN;
                    } else {
                            --$bucket;
                            my $entry = shift @$queue;
                            $entry->{code}(@{$entry->{args}});
                    }
            }
    }

Enfin, l'état du bucket est sauvegardé dans la structure de données $ conn (en fait un peu plus tard dans la méthode; il calcule d'abord combien de temps il aura plus de travail)

    # Save status.
    $conn->{fujiko_limit_bucket} = $bucket;
    $conn->{fujiko_limit_lasttx} = $start_time;

Comme vous pouvez le voir, le code de gestion du bucket est très petit - environ quatre lignes. Le reste du code est la gestion de la file d'attente prioritaire. Le bot a des files d'attente prioritaires afin que, par exemple, quelqu'un qui discute avec lui ne puisse pas l'empêcher de faire ses importantes tâches de kick / ban.

pour bloquer le traitement jusqu'à ce que le message puisse être envoyé, mettant ainsi d'autres messages en file d'attente, la belle solution d'antti peut également être modifiée comme ceci:

rate = 5.0; // unit: messages
per  = 8.0; // unit: seconds
allowance = rate; // unit: messages
last_check = now(); // floating-point, e.g. usec accuracy. Unit: seconds

when (message_received):
  current = now();
  time_passed = current - last_check;
  last_check = current;
  allowance += time_passed * (rate / per);
  if (allowance > rate):
    allowance = rate; // throttle
  if (allowance < 1.0):
    time.sleep( (1-allowance) * (per/rate))
    forward_message();
    allowance = 0.0;
  else:
    forward_message();
    allowance -= 1.0;

il attend juste qu'il y ait suffisamment de marge pour envoyer le message. pour ne pas commencer avec deux fois le taux, la tolérance peut également être initialisée à 0.

Conservez l'heure à laquelle les cinq dernières lignes ont été envoyées. Maintenez les messages en file d'attente jusqu'à ce que le cinquième message le plus récent (s'il existe) soit au moins 8 secondes dans le passé (avec last_five comme tableau de fois):

now = time.time()
if len(last_five) == 0 or (now - last_five[-1]) >= 8.0:
    last_five.insert(0, now)
    send_message(msg)
if len(last_five) > 5:
    last_five.pop()

Une solution consiste à attacher un horodatage à chaque élément de la file d'attente et à supprimer l'élément au bout de 8 secondes. Vous pouvez effectuer cette vérification à chaque fois que la file d'attente est ajoutée.

Cela ne fonctionne que si vous limitez la taille de la file d'attente à 5 et annulez tout ajout alors que la file d'attente est pleine.

Si quelqu'un est toujours intéressé, j'utilise cette classe appelable simple en conjonction avec un stockage de valeur de clé LRU chronométré pour limiter le taux de demande par IP. Utilise un deque, mais peut être réécrit pour être utilisé avec une liste à la place.

from collections import deque
import time


class RateLimiter:
    def __init__(self, maxRate=5, timeUnit=1):
        self.timeUnit = timeUnit
        self.deque = deque(maxlen=maxRate)

    def __call__(self):
        if self.deque.maxlen == len(self.deque):
            cTime = time.time()
            if cTime - self.deque[0] > self.timeUnit:
                self.deque.append(cTime)
                return False
            else:
                return True
        self.deque.append(time.time())
        return False

r = RateLimiter()
for i in range(0,100):
    time.sleep(0.1)
    print(i, "block" if r() else "pass")

Juste une implémentation python d'un code de réponse acceptée.

import time

class Object(object):
    pass

def get_throttler(rate, per):
    scope = Object()
    scope.allowance = rate
    scope.last_check = time.time()
    def throttler(fn):
        current = time.time()
        time_passed = current - scope.last_check;
        scope.last_check = current;
        scope.allowance = scope.allowance + time_passed * (rate / per)
        if (scope.allowance > rate):
          scope.allowance = rate
        if (scope.allowance < 1):
          pass
        else:
          fn()
          scope.allowance = scope.allowance - 1
    return throttler

Que dis-tu de ça:

long check_time = System.currentTimeMillis();
int msgs_sent_count = 0;

private boolean isRateLimited(int msgs_per_sec) {
    if (System.currentTimeMillis() - check_time > 1000) {
        check_time = System.currentTimeMillis();
        msgs_sent_count = 0;
    }

    if (msgs_sent_count > (msgs_per_sec - 1)) {
        return true;
    } else {
        msgs_sent_count++;
    }

    return false;
}

J'avais besoin d'une variante de Scala. C'est ici:

case class Limiter[-A, +B](callsPerSecond: (Double, Double), f: A ⇒ B) extends (A ⇒ B) {

  import Thread.sleep
  private def now = System.currentTimeMillis / 1000.0
  private val (calls, sec) = callsPerSecond
  private var allowance  = 1.0
  private var last = now

  def apply(a: A): B = {
    synchronized {
      val t = now
      val delta_t = t - last
      last = t
      allowance += delta_t * (calls / sec)
      if (allowance > calls)
        allowance = calls
      if (allowance < 1d) {
        sleep(((1 - allowance) * (sec / calls) * 1000d).toLong)
      }
      allowance -= 1
    }
    f(a)
  }

}

Voici comment il peut être utilisé:

val f = Limiter((5d, 8d), { 
  _: Unit ⇒ 
    println(System.currentTimeMillis) 
})
while(true){f(())}