iterqueue_speed_test.py

Profiling the iterqueue extended iterator classes

import sys, os, itertools
from timeit import Timer
import iterqueue
from iterqueue_test import wrappers, peekables, pushables, \
     state_reporters, iqueues


def print_iterator_lists():
    print "Wrappers"
    print " ","\n  ".join(wrapper.__name__ for wrapper in wrappers)
    print "Peekables"
    print " ","\n  ".join(peekable.__name__ for peekable in peekables)
    print "Pushables"
    print " ","\n  ".join(pushable.__name__ for pushable in pushables)
    print "State Reporters"
    print " ","\n  ".join(state_reporter.__name__
                     for state_reporter in state_reporters)
    print "Iterator Queues"
    print " ","\n  ".join(iqueue.__name__ for iqueue in iqueues)


#print_iterator_lists()

use cases (benchmarks)

def loop(iterator):
    """baseline: empty `for` loop"""
    for _ in iterator:
        pass

def peek_in_loop(iterator):
    """peek in every loop"""
    for _ in iterator:
        try:
            iterator.peek()
        except StopIteration:
            pass

def peek_before_loop(iterator):
    """peek at first value once, then loop"""
    try:
        iterator.peek()
    except StopIteration:
        pass
    for _ in iterator:
        pass

def bool_in_loop(iterator):
    """test for values in every loop"""
    for _ in iterator:
        bool(iterator)

def bool_before_loop(iterator):
    """test for values once, then loop"""
    bool(iterator)
    for _ in iterator:
        pass


def time_benchmark(fun, wrappers, iterator):
    """profile benchmark `fun` with `iterator` wrapped in `wrappers`"""

    print fun.__doc__, "(%s)"%iterator
    setup = "import iterqueue_speed_test\nimport iterqueue"
    benchmark = "iterqueue_speed_test.%s(iterqueue.%s(%s))"
    stmts = [benchmark%(fun.__name__, wrapper.__name__, iterator)
             for wrapper in wrappers]
    timers = [Timer(stmt=stmt, setup=setup) for stmt in stmts]

    t_i = [min(timer.repeat(number=1, repeat=3)) for timer in timers]

    results = ["%.5f s   %s"%(t, wrapper.__name__)
               for t, wrapper in zip(t_i, wrappers)]
    results.sort()
    print "\n".join(results)

time_benchmark(loop, iterqueue.XIter, xrange(1000))

time_benchmark(loop, wrappers, xrange(1000))
print
time_benchmark(peek_before_loop, peekables, xrange(1000))
print
time_benchmark(peek_in_loop, peekables, xrange(1000))
print
time_benchmark(bool_before_loop, state_reporters, xrange(1000))
print
time_benchmark(bool_in_loop, state_reporters, xrange(1000))