¿Qué es un buen algoritmo de limitación de velocidad?

StackOverflow https://stackoverflow.com/questions/667508

  •  21-08-2019
  •  | 
  •  

Pregunta

Yo podría utilizar algunos pseudo-código, o mejor, Python. Estoy tratando de poner en práctica una cola limitante de la velocidad para un robot Python IRC, y que funciona parcialmente, pero si alguien provoca menos mensajes que el límite (por ejemplo, límite de velocidad es de 5 mensajes por 8 segundos, y la persona que dispara sólo 4), y el siguiente disparo es durante los 8 segundos (por ejemplo, 16 segundos más tarde), el bot envía el mensaje, pero la cola se llena y el bot espera 8 segundos, a pesar de que no es necesario ya que ha transcurrido el segundo período de 8.

¿Fue útil?

Solución

Aquí el más simple algoritmo , si quieres sólo para dejar mensajes cuando lleguen demasiado rápido ( en lugar de hacer cola ellos, lo cual tiene sentido porque la cola podría obtener arbitrariamente 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;

No hay estructuras de datos, temporizadores, etc., en esta solución y funciona limpiamente :) Para ver esto, 'asignación' crece a una velocidad de 5/8 unidades por segundo como máximo, es decir, un máximo de cinco unidades por ocho segundos. Cada mensaje que se reenvía disminuye en una unidad, por lo que no puede enviar más de cinco mensajes por cada ocho segundos.

Tenga en cuenta que rate debe ser un entero, es decir, sin no igual a cero decimal, o el algoritmo no funcionará correctamente (tasa real no será rate/per). P.ej. rate=0.5; per=1.0; no funciona porque allowance nunca crecerá a 1,0. Pero rate=1.0; per=2.0; funciona bien.

Otros consejos

Utilice este decorador @RateLimited (ratepersec) antes de la función que encola.

Básicamente, se comprueba si seg 1 / frecuencia han pasado desde la última vez y si no, espera el resto del tiempo, de lo contrario no espera. Esta eficacia se limita a la tasa / seg. El decorador puede aplicarse a cualquier función que desee tasa limitada.

En su caso, si quieres un máximo de 5 mensajes por cada 8 segundos, utilice @RateLimited (0.625) antes de que su función 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)

A Token Bucket es bastante simple de implementar.

Comience con un cubo con 5 fichas.

Cada 5/8 segundos:. Si el cubo tiene menos de 5 fichas, agrega uno

Cada vez que desee enviar un mensaje: Si el cubo tiene ≥1 manera, tomar una ficha y enviar el mensaje. De lo contrario, esperar / dejar el mensaje / lo que sea.

(obviamente, en el código actual, tendrá que utilizar un contador de números enteros en lugar de fichas reales y se puede optimizar el cada paso 5 / 8s mediante el almacenamiento de las marcas de tiempo)

La lectura de la pregunta de nuevo, si el límite de velocidad se restablece plenamente cada 8 segundos, entonces aquí es una modificación:

Comience con una marca de tiempo, last_send, a la vez hace mucho tiempo (por ejemplo, en la época). Además, comenzar con la misma cubeta de 5 token.

Huelga de la regla cada 5/8 segundos.

Cada vez que envíe un mensaje: En primer lugar, comprobar si <=> ≥ hace 8 segundos. Si es así, llenar el cubo (en posición 5 tokens). En segundo lugar, si hay fichas en el cubo, enviar el mensaje (de lo contrario, gota / espera / etc.). En tercer lugar, establecer <=> ahora.

Esto debería funcionar para ese escenario.

De hecho, he escrito un bot de IRC usando una estrategia como esta (primera aproximación). Su en Perl, Python no, pero aquí hay un código para ilustrar:

La primera parte aquí se encarga de añadir fichas al cubo. Se puede ver la optimización de la adición de fichas en función del tiempo (segunda a la última línea) y luego la última línea de abrazaderas contenidos de cubo al máximo (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 es una estructura de datos que se pasa alrededor. Esto está dentro de un método que se ejecuta de forma rutinaria (se calcula que la próxima vez que tendrá algo que ver, y duerme bien tanto tiempo o hasta que se pone el tráfico de red). La siguiente parte del método maneja el envío. Es bastante complicado, ya que los mensajes han prioridades asociado a ellos. 

    # 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] = [];

Esta es la primera cola, que está dirigido a toda costa. Incluso si se pone nuestra conexión mataron por las inundaciones. Se utiliza para cosas muy importantes, como respuesta a PING del servidor. A continuación, el resto de las colas: 

    # 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}});
                    }
            }
    }

Finalmente, el estado de la cubeta se guarda de nuevo a la estructura de datos $ conn (en realidad un poco más tarde en el método, sino que primero calcula cuán pronto tendrá más trabajo) 

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

Como se puede ver, el código de manejo real de la cuchara es muy pequeña - alrededor de cuatro líneas. El resto del código es la manipulación cola de prioridad. El robot tiene colas de prioridad de manera que, por ejemplo, alguien que charla con él no puede impedirlo sus importantes funciones kick / ban.

bloquear el procesamiento hasta que el mensaje puede ser enviado, cola de este modo hasta más mensajes, solución bonitos de antti también puede modificarse como sigue: 

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;

que sólo espera hasta que haya suficiente provisión está ahí para enviar el mensaje. a no empezar con el doble de la tasa, teniendo en cuenta también puede inicializado con 0.

Mantener el momento en que se enviaron los últimos cinco líneas. Mantenga los mensajes en cola hasta el momento en el quinto más reciente-mensaje (si existe) es un menos 8 segundos en el pasado (con last_five como una matriz de veces): 

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()

Una solución es para adjuntar un sello de tiempo a cada elemento de cola y para descartar el artículo después de 8 segundos han pasado. Puede realizar esta comprobación cada vez que se agrega a la cola.

Esto sólo funciona si se limita el tamaño de la cola a 5 y descartar cualquier adición, mientras que la cola está llena.

Si alguien sigue interesado, que utilizan esta clase exigible simple en conjunción con un LRU almacenamiento de valor de clave programada para limitar tasas de solicitudes por IP. Utiliza una doble cola, pero puede reescrito para ser utilizado con una lista en su lugar. 

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")

Sólo una aplicación de pitón de un código de respuesta aceptada. 

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

¿Qué tal esto: 

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;
}

que necesitaba una variación en la Scala. Aquí está: 

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)
  }

}

Así es cómo se puede utilizar: 

val f = Limiter((5d, 8d), { 
  _: Unit ⇒ 
    println(System.currentTimeMillis) 
})
while(true){f(())}
Licenciado bajo: CC-BY-SA con atribución
No afiliado a StackOverflow
scroll top