Riak benchmark, output: avg 0.0346335514806 p90 0.0778579711914 p99 0.190546035767

riak-benchmark.py

from time import time
from twisted.web import client, error as web_error
from twisted.python import failure
from twisted.internet import task, defer
from cStringIO import StringIO
import gzip
import random

port = 8098

def get(host, bucket, key, timeout=0.5):

    def handle_result(result):
        return None

    def build_url(host):
        return str(url)

    def on_404(failure):
        if failure.check(web_error.Error) and failure.value.status == 404:
            return None
        return failure

    url = 'http://%s:%s/riak/%s/%s' % (host, port, bucket, key)
    d = client.getPage(url , timeout=timeout)
    d.addCallbacks(handle_result, on_404)
    return d


def benchmark():
    request_count = 2000
    all_ids = open('../us_users.txt').read().split('\n')
    ids = random.sample(all_ids, request_count)
    host = 'riak-89.production.wrops.net'
    timeout = 20
    concurrent_workers = 1
    times = []

    def iter():
        def log_result(r):
            print "Riak responded in %s s" % r
            return r
        def handle_failure(f):
            if f.type == defer.TimeoutError:
                print "Timed out!"
                return timeout
            print f.value
            return None
        for id in ids:
            start = time()
            d = get(host, 'friendships', str(id), timeout=timeout)
            d.addCallbacks(lambda ign: time()-start, handle_failure)
            d.addCallback(log_result)
            d.addCallback(lambda v: times.append(v))
            yield d

    ds = []
    it = iter()
    for i in range(concurrent_workers):
        ds.append(task.coiterate(it))

    def build_result(ign):
        filtered = filter(lambda v: v, times)
        print "Failures ", (len(times)-len(filtered))
        return filtered

    d = defer.gatherResults(ds)
    d.addCallback(build_result)
    return d

if __name__ == "__main__":
    from twisted.internet import reactor
    def got_result(r):
        ts = sorted(r)
        print "avg", sum(ts)/len(ts)
        print "p90", min(ts[int(len(ts)*0.9):])
        print "p99", min(ts[int(len(ts)*0.99):])
        reactor.stop()

    d = benchmark()
    d.addCallback(got_result)
    reactor.run()