Qu'est-ce qu'un bon taux limitant l'algorithme?
https://stackoverflow.com/questions/667508
-
21-08-2019 - |
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
RussianQuestion
Je pourrais utiliser un pseudo-code, ou mieux, Python. Je suis en train de mettre en œuvre une file d'attente de limitation de vitesse pour un bot IRC Python, et il fonctionne en partie, mais si quelqu'un déclenche moins de messages que la limite (par exemple, la limite de vitesse est de 5 messages par 8 secondes, et la personne ne se déclenche que 4), et le déclenchement suivant est au cours des 8 secondes (par exemple, 16 secondes plus tard), le robot envoie le message, mais la file d'attente est pleine et le bot attend 8 secondes, même si ce n'est pas nécessaire puisque la période 8 secondes est écoulé.
La solution
Voici le simple algorithme , si vous voulez juste déposer des messages quand ils arrivent trop vite ( au lieu de les faire la queue, ce qui est logique car la file d'attente peut obtenir arbitrairement grand):
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 datastructures, minuteries, etc. dans cette solution et il fonctionne proprement :) Pour voir cela, « indemnité » croît à la vitesse des unités 5/8 par seconde au maximum, à savoir au plus cinq unités par huit secondes. Chaque message qui est transmis d'une unité déduit, donc vous ne pouvez pas envoyer plus de cinq messages par toutes les huit secondes.
Notez que doit être un rate entier, à savoir sans partie décimale non nulle, ou l'algorithme ne fonctionne pas correctement (taux réel ne sera pas rate/per). Par exemple. Ne fonctionne pas rate=0.5; per=1.0; parce que ne se développera allowance à 1,0. Mais fonctionne très bien rate=1.0; per=2.0;.
Autres conseils
Utilisez ce décorateur @RateLimited (ratepersec) avant votre fonction enqueues.
En gros, ce vérifie si 1 / sec de taux se sont écoulés depuis la dernière fois et sinon, attend le reste du temps, sinon il n'attend pas. Cela vous limite efficacement à taux / sec. Le décorateur peut être appliqué à toutes les fonctions que vous voulez vitesse limitée.
Dans votre cas, si vous voulez un maximum de 5 messages par 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 jeton seau est assez simple à mettre en œuvre.
Commencez avec un seau avec 5 jetons.
Toutes les 5/8 secondes. Si le seau a moins de 5 jetons, ajoutez un
Chaque fois que vous voulez envoyer un message: Si le seau a ≥1 jeton, prenez un jeton et d'envoyer le message. Sinon, attendez / déposer le message / whatever.
(évidemment, dans le code réel, vous utiliseriez un compteur entier au lieu de jetons réels et vous pouvez optimiser la chaque étape 5 / 8s en stockant horodatages)
La lecture de la question à nouveau, si la limite de vitesse est remis à zéro complètement chaque 8 secondes, puis est une modification ici:
Commencez par un horodatage, last_send, à un moment il y a longtemps (par exemple, à l'époque). En outre, commencer avec le même seau de 5 jetons.
Frappez la seconde règle toutes les 5/8.
Chaque fois que vous envoyez un message: Tout d'abord, vérifiez si ≥ il y a 8 secondes <=>. Si oui, remplir le seau (régler sur 5 jetons). En second lieu, s'il y a des jetons dans le seau, envoyer le message (sinon, baisse / attente / etc.). Troisièmement, mettre maintenant <=>.
Cela devrait fonctionner pour ce scénario.
Je suis en fait écrit un bot IRC en utilisant une stratégie comme celui-ci (la première approche). Son en Perl, Python pas, mais voici un code pour illustrer:
La première partie traite ici d'ajouter des jetons dans le seau. Vous pouvez voir l'optimisation d'ajouter des jetons en fonction du temps (2 à la dernière ligne), puis la dernière ligne pinces 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 passé autour. Ceci est à l'intérieur d'une méthode qui fonctionne régulièrement (il calcule quand la prochaine fois qu'il aura quelque chose à faire, et peut accueillir soit si longtemps ou jusqu'à ce qu'il obtienne le trafic réseau). La partie suivante de la méthode gère l'envoi. Il est assez compliqué, parce que les messages ont des priorités qui leur sont associé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é, peu importe quoi. Même si elle obtient notre connexion tué pour les inondations. Utilisé pour des choses extrêmement importantes, comme répondant à la 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 godet est enregistré en arrière à la structure de données conn $ (en fait un peu plus tard dans la méthode, il calcule d'abord combien de temps il faudra 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 seau réel est très faible - environ quatre lignes. Le reste du code est la gestion des files d'attente prioritaire. Le bot a des files d'attente prioritaires de sorte que par exemple, quelqu'un bavarder avec elle ne peut pas l'empêcher de faire ses importants kick / ban fonctions.
pour bloquer le traitement jusqu'à ce que le message peut être envoyé, ainsi faire la queue d'autres messages, belle solution de antti peut également être modifiée comme suit:
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 jusqu'à ce que l'allocation est-il suffisant pour envoyer le message. de ne pas commencer par deux fois le taux, l'allocation peut également initialisé avec 0.
Gardez le temps que les cinq dernières lignes ont été envoyées. Tenir les messages en file d'attente jusqu'à ce que le temps le cinquième le plus récent message (si elle existe) est au moins 8 secondes dans le passé (avec last_five comme un 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 à fixer une estampille temporelle à chaque élément de file d'attente et d'écarter l'élément après 8 secondes se sont écoulées. Vous pouvez effectuer cette vérification à chaque fois que la file d'attente est ajouté à.
Cela ne fonctionne que si vous limitez la taille de la file d'attente à 5 et défaussez tout ajout alors que la file d'attente est pleine.
Si quelqu'un toujours intéressé, j'utiliser cette classe simple appelable en conjonction avec un stockage chronométré LRU valeur clé pour limiter le taux de demande par IP. Utilise un deque, mais peut réécrite 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 diriez-vous ceci:
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;
}
Je avais besoin d'une variation de Scala. Ici, il 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)
}
}
Voici comment il peut être utilisé:
val f = Limiter((5d, 8d), {
_: Unit ⇒
println(System.currentTimeMillis)
})
while(true){f(())}
