riak benchmark multiget

riak_multiget.py

"""
Results are in:
user / system / total / (real)

Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 8
Keys: 50
Rehearsal -------------------------------------------------
single thread     0.020000     0.000000 (     0.050000 )
multi thread     0.020000     0.010000 (     0.040000 )
-----------------------------------------------------------

Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 16
Keys: 50
Rehearsal -------------------------------------------------
single thread     0.010000     0.000000 (     0.060000 )
multi thread     0.020000     0.020000 (     0.040000 )
-----------------------------------------------------------


Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 8
Keys: 100
Rehearsal -------------------------------------------------
single thread     0.030000     0.000000 (     0.110000 )
multi thread     0.040000     0.010000 (     0.050000 )
-----------------------------------------------------------


Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 16
Keys: 100
Rehearsal -------------------------------------------------
single thread     0.030000     0.000000 (     0.100000 )
multi thread     0.040000     0.020000 (     0.060000 )
-----------------------------------------------------------


Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 8
Keys: 1000
Rehearsal -------------------------------------------------
single thread     0.260000     0.030000 (     0.950000 )
multi thread     0.340000     0.100000 (     0.450000 )
-----------------------------------------------------------

Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 16
Keys: 1000
Rehearsal -------------------------------------------------
single thread     0.220000     0.020000 (     0.700000 )
multi thread     0.400000     0.040000 (     0.460000 )
-----------------------------------------------------------

Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 8
Keys: 10000
Rehearsal -------------------------------------------------
single thread     2.780000     0.220000 (     9.120000 )
multi thread     3.630000     0.840000 (     4.560000 )
-----------------------------------------------------------

Riak multiget:
Riak Nodes: 5
CPUs: 8
Threads: 16
Keys: 10000
Rehearsal -------------------------------------------------
single thread     2.640000     0.220000 (     8.960000 )
multi thread     3.580000     0.970000 (     4.420000 )
-----------------------------------------------------------

"""


# -*- coding: utf-8 -*-

from contextlib import closing
import sys
import timeit
import functools
from multiprocessing import cpu_count
import riak.benchmark as benchmark
import riak

NODES = [
    {'host': '127.0.0.1', 'pb_port': 10017},
    {'host': '127.0.0.1', 'pb_port': 10027},
    {'host': '127.0.0.1', 'pb_port': 10037},
    {'host': '127.0.0.1', 'pb_port': 10047},
    {'host': '127.0.0.1', 'pb_port': 10057},
]

NODE_LEN = len(NODES)
POOL_SIZE = 16

client = riak.RiakClient(nodes=NODES, protocol='pbc', multiget_pool_size=POOL_SIZE)
bucket_type = 'default'
bucket_name = 'test'
bucket = client.bucket(bucket_name)



def non_threaded(keys):
    for k in keys:
        bucket.get(k)


def threaded(keys):
    bucket.multiget(keys)


def get_data(length):
    length = int(length)
    return [{'key': str(i), 'data': {'name': 'Esteban', 'last_name': 'nitely'}} for i in xrange(length)]


def populate(items):
    for item in items:
        bucket\
            .new(item['key'], data=item['data'])\
            .store(w=3)


def cleanup():
    with closing(client.stream_keys(bucket)) as keys:
        for key_list in keys:
            for key in key_list:
                bucket\
                    .get(key)\
                    .delete(w=3)


def run():
    data_len = sys.argv[1]
    items = get_data(data_len)
    keys = [item['key'] for item in items]

    populate(items)

    #print timeit.timeit(functools.partial(non_threaded, keys), number=100)
    #print timeit.timeit(functools.partial(threaded, keys), number=100)

    for b in benchmark.measure_with_rehearsal():
        with b.report('single thread'):
            non_threaded(keys)

        with b.report('multi thread'):
            threaded(keys)

    cleanup()


if __name__ == "__main__":
    print "Riak multiget:"
    print "Riak Nodes: {0}".format(NODE_LEN)
    print "CPUs: {0}".format(cpu_count())
    print "Threads: {0}".format(POOL_SIZE)
    print "Keys: {0}".format(sys.argv[1])
    run()
    print "ok"