Funktionen benchmarken mit timeout.

[akis.kapo]

Hi all,

kennt ihr das allgemeine Problem, ihr habt 3-4 Ideen, wie ihr eine Funktion implementieren könntet.

Die eine ist pythonisch und sieht schön kurz und elegant aus, die andere sieht mehr nach C aus und hässlich,
dann kommen noch verschiedene Variationen hinzu, die einen mehr oder weniger verständlich, die anderen mehr oder weniger optimiert.

Am Ende stellt sich aber die Frage, welche ist schneller?

Normalerweise benchmarke ich meine Funktionen selber, a la ...

Code: Alles auswählen

import time

def func_v1():
    do something ...

def func_v2():
    do something, slightly different ...

iterations=10000

v1_start = time.time()
for _ in xrange(iterations):
    func_v1()
print "func_v1() took {0:f} seconds for {1:d} iterations.".format(time.time() - v1_start, iterations)

v2_start = time.time()
for _ in xrange(iterations):
    func_v2()
print "func_v2() took {0:f} seconds for {1:d} iterations.".format(time.time() - v2_start, iterations)

Das ist mega-käsig, ich weiss. Zudem bin ich gerade eben durch suchen nach "timeout" im Forum auf das "timeit" Modul gestoßen, welches mein Vorgehen oben komplett ersetzt.

Aber deswegen bin ich nicht hier, denn ich suche nach eine ganz anderen Lösung:

Also ein allgemeines Problem beim Code ist ja, dass man kein echtes Gefühl hat für die absolute Ausführungsgeschwindigkeit.
D.h. selbst wenn man sich zum benchmarken für die Methodik "wieviele Sekunden brauchst du für x Durchgänge" entscheidet,
muss man trotzdem zuvor erstmal ein paar Durchgänge ausprobieren, um überhaupt ein Gefühl dafür zu bekommen, wie man
die Anzahl der Iterationen x wählt für jede einzelne Funktion, denn manche brauchen Millisekunden, manche Minuten, andere fast ne Stunde.

Aus diesem Grund möchte ich es eigentlich umgekehrt haben: "wieviele Iterationen hast du für eine fixe Zeit x geschafft."

Dazu bräuchte ich eine Möglichkeit, eine Funktion auszuführen, jeden vollständigen Durchlauf zu zählen (counter) und wenn das definierte Timeout,
zum Beispiel 10 Sekunden, erreicht ist, soll die (äussere) Benchmarkfunktionen sofort abbrechen und die ge-benchmarkte Funktionen killen.

Fantasiecode:

Code: Alles auswählen

def benchthis(func, timeout):
    count = 0
    t = Timeout.set_timeout(timeout)
    try:
        func()
        count += 1
    except t.timeout_reached(), e:
        pass
    print "completet {0:d} iterations in {1:f} seconds.".format(count, timeout)

Und ja, bitte nicht auf eventlet verweisen, das hab ich schon durch googlen zufällig entdeckt, aber ich würde gerne wissen, wie man das mit Python-boardmitteln am besten macht.

Solange ich keine gescheite Lösung habe nutze ich ne total jämmerliche Fallbacklösung, die keine Garantie dafür gibt, tatsächlich nach dem timeout abzubrechen.

Code: Alles auswählen

def benchme(func, timeout):
    c = 0
    start = time.time()
    while timeout < (time.time() - start):
        func()
        c += 1
    return c, timeout

Wenn func() hier mehr als eine Stunde dauert, läuft genau ein func() komplett durch, und nach einer Stunde erst hört benchme auf und gibt als counter 1 zurück. Toll.

Also ich hoffe ich habe das Problem klar genug dargestellt und bin froh um Vorschläge, wie man das am besten mit Python-Boardmitteln implementieren könnte.

Im Voraus vielen Dank für Eure Antworten und Beiträge.

[BlackJack]

@akis.kapo: Ich denke das einzige allgemeine was in den meisten Fällen funktioniert und mit der Standardbibliothek auskommt ist das `multiprocessing`-Modul weil man Prozesse, im Gegensatz zu Threads, hart/bedingungslos beenden kann. Wenn man sich auf Unix beschränkt käme noch das `signal`-Modul in Frage.

Mit Multiprocessing würde ich wohl einen Prozess starten der dann ein von `Thread` abgeleitetes Objekt als Thread startet und die Funktion in einer Schleife ausführt und die Durchgänge in einem Attribut zählt. Im Hauptthread dann ein `time.sleep()` mit dem gewünschten Timeout, was danach das Attribut abfragt und dem Prozess-„Aufrufer” mitteilt, der darauf hin den Process killt und damit die Funktion im Thread mit beendet.

[akis.kapo]

EDIT: hab grad gemerkt das ist dasselbe in grün, was BlackJack zuvor gepostet hat.
Aber ich denke es wird wohl auf ne Kombination aus beidem, multiprocessing und thead/threading hinauskommen.
Denn zum einen muss man ne Funktion an einem Prozess binden, um diese gezielt töten zu können,
zum anderen muss das ganze nebenläufig geschehen, also in einem echten Thread gekapselt sein.

Ich les grad die Libref "multiprocessing, threading, thead und subprocess", um mir ein Bild zu machen, wie man das am besten zusammenhackt...

EDIT: mir blüht gerade, bei so mini-Funktionen im Millisekunden-Bereich wäre der Ansatz über Prozesse evtl. konterproduktiv und am Ende nichtsaussagend.
Man will ja nicht den OS overhead für's forken eines neuen Prozesses in die Mikrotransaktion einer internen Funktion mit einfliessen lassen. Das verfälscht ja die Ergebnisse beliebig.
Ich überlege grad, wie man das am besten macht, genau einen Prozess zu generieren, der die Funktionen beliebig oft aufruft, statt einen Prozess pro Funktionsaufruf zu forken.

[akis.kapo]

Ok, ich bin schon weiter, aber irgendwie auch wieder nicht.

Wieso tut das hier nicht, wie ich will? Wo ist der flaw?

Code: Alles auswählen

import multiprocessing, threading

class benchme:
    def __init__(self, func, timeout=10):
        self.iterations = 0
        self.func = func
        self.proc = multiprocessing.Process(target=self.repeat)
        self.timeout = timeout
        self.timer = threading.Timer(self.timeout, self.proc.terminate)

    def repeat(self):
        while True:
            self.func()
            self.iterations += 1

    def evaluate(self):
        self.proc.start()
        self.timer.start()
        self.proc.join()
        print "{0:d} iterations in {1:.2f} seconds".format(self.iterations, self.timeout)
        return self.iterations

Testdatei:

Code: Alles auswählen

from benchme import *

def f1():
    for _ in xrange(1000):
        pass

def f2():
    for _ in xrange(10000000):
        pass

b1 = benchme(func=f1)
i1 = b1.evaluate()

b2 = benchme(func=f2)
i2 = b2.evaluate()

Output:

Code: Alles auswählen

$ ./benchme_sample.py 
0 iterations in 10.00 seconds
0 iterations in 10.00 seconds

Also irgendwas mach ich falsch... kann mich jemand in die richtige Richtung tippen?

[BlackJack]

@akis.kapo: Nur geraten, aber wenn `self.repeat()` in einem anderen *Prozess* ausgeführt wird, dann muss ja das Objekt auf dem die Methode ausgeführt wird serialisiert und an den anderen Prozess übertragen werden. In dem anderen Prozess gibt es dann eine Kopie von dem Objekt und *dort* wird das Attribut hochgezählt, was nichts mit dem Attribut zu tun hat welches im startenden Prozess auf dem Originalobjekt existiert.

Deshalb passiert in meiner Beschreibung auch alles in dem externen Prozess, also erst den Prozess erstellen und darin den Thread zum berechnen starten. Dann das Ergebnis nach der Timeout-Zeit zurückgeben und den Aufrufer, also den ersten Prozess den gestarteten Prozess beenden lassen. Man muss bei `multiprocessing` noch mehr als bei `threading` darauf achten was in welchem Ausführungsstrang passiert, weil bei `multiprocessing` kein gemeinsamer Adressraum existiert und implizit Kopien von Werten erstellt werden.

Zur Fehlersuche kann es hier beispielsweise hilfreich sein sich an strategischen Stellen den Wert von `self` oder `id(self)` ausgeben zu lassen, ob die Annahme über das selbe Objekt tatsächlich das *selbe* Objekt betrifft.

[akis.kapo]

Ich hab's zwischendurch hiermit probiert, jedoch weiterhin ohne Ergebnis:

Code: Alles auswählen

def repeat(self):
    try:
        while True:
            self.func()
            self.iterations += 1
    finally:
        print self.iterations

Leider versteht .terminate() kein Spaß, da hilft auch kein finally Block...

[akis.kapo]

Hier ist eine Lösung, siehe http://docs.python.org/2/library/multip ... -processes:

Code: Alles auswählen

import multiprocessing, threading

class benchme:
    def __init__(self, func, timeout=3):
        self.iterations = multiprocessing.Value('L', 0)
        self.func = func
        self.proc = multiprocessing.Process(target=self.repeat, args=(self.iterations,))
        self.timeout = timeout
        self.timer = threading.Timer(self.timeout, self.proc.terminate)

    def repeat(self, iters):
        while True:
            self.func()
            iters.value += 1

    def start(self):
        self.proc.start()
        self.timer.start()
        self.proc.join()

    def display(self):
        print "{0:g} iterations per seconds.".format(self.get_ips())

    def get_results(self):
        return (self.iterations.value, self.timeout)

    def get_iterations(self):
        i, _ = self.get_results()
        return i

    def get_ips(self):
        i, s = self.get_results()
        return i / float(s)

    def evaluate(self):
        print "executing {0:s}() with timeout {1:.2f} seconds... ".format(str(self.func.__name__), self.timeout)
        self.start()
        self.display()

Testskript:

Code: Alles auswählen

from benchme import *

def f1():
    for _ in xrange(1000):
        pass

def f2():
    for _ in xrange(10000000):
        pass

b1 = benchme(func=f1)
i1 = b1.evaluate()

b2 = benchme(func=f2)
i2 = b2.evaluate()

Kommandozeile:

Code: Alles auswählen

$ ./benchme_sample.py 
executing f1 with timeout 3.00 seconds...
39637.7 iterations per seconds.
executing f2 with timeout 3.00 seconds...
4.66667 iterations per seconds.

Scheint als wenn multiprocessing.Value hier unser Freund ist. Einer, von mehreren möglichen...
Man kann auch Queues und Pipes verwenden, aber irgendwie hat mich Value hier für diesen Verwendungszweck am meisten angesprochen.

Danke @BlackJack.

EDIT: bissle Schnick-Schnack in den Code hinzugefügt.