python-hs-benchmark /

#!/usr/bin/env python

import test
import time
from multiprocessing import Pool
import memcache
import pylibmc
import handlersocket
import handlersocket_innodb

class WritableObject:
    def __init__(self):
    def write(self, string):
    def flush(self):

def base_test(test_no, backend):
    global options
    test.options = options
    b = test.Benchmark(backend, options, 'test_%d' % test_no)
    return b.total_time

# You cannot pass modules or functions to the processes
def memcache_test(test_no):
    return base_test(test_no, memcache)

def pylibmc_test(test_no):
    return base_test(test_no, pylibmc)

def handlersocket_test(test_no):
    return base_test(test_no, handlersocket)

def handlersocket_innodb_test(test_no):
    return base_test(test_no, handlersocket_innodb)

if __name__ == '__main__':
    from optparse import OptionParser
    parser = OptionParser()

    parser.add_option('-a', '--server-address', dest='server_address',
            help="address:port of memcached [default:]")
    parser.add_option('-o', '--output-csv', dest='csv',
            help="save test results into given csv file")

    parser.add_option('-n', '--num-tests', dest='num_tests', type='int',
            help="repeat counts of each test [default: 1000]")

    parser.add_option('-f', '--min-processes', dest='min_processes', type='int',
            help="amount of processes to start with [default: 1]")

    parser.add_option('-t', '--max-processes', dest='max_processes', type='int',
            help="maximal amount of processes to run [default: 5]")

    parser.add_option('-v', '--verbose', dest='verbose',
            action='store_true', default=False,
            help="show traceback infomation if a test fails")

    global options
    options, args = parser.parse_args()
    options.quiet = True
    pool = Pool(processes=options.max_processes)
    grand_total = []

    for func in [memcache_test, pylibmc_test, handlersocket_test, handlersocket_innodb_test,]:
        line_in_total = [func.__name__]
        print "=" * 20
        print "Testing %s" % func.__name__
        print "=" * 20

        for i in range(options.min_processes, options.max_processes + 1):
            start_time = time.time()
            timings =, range(i))
            end_time = time.time()

            print "Overall time for %d processes is %f seconds" % (i, end_time - start_time)
            avg = sum(timings, 0.0) / len(timings)
            print "Average clean (without setup/teardown) execution time is %f seconds per thread\n" % avg
    if options.csv:
        import csv
        writer = csv.writer(open(options.csv, "wb"))
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.