ep-engine strategy performance test thing

perftest.py

#!/usr/bin/env python

import os
import sys
import time
import glob
import string
import random
import subprocess

import mc_bin_client

CHARSET = string.ascii_letters + string.digits

NUM_KEYS = 3000000
KEYLEN = 8
VALLEN = 2048
VBUCKETS = 1024
UP_ITERS = 7

NUM_ATTEMPTS = 7

def debug(x, fd=sys.stdout):
    fd.write("# %s\n" % x)
    fd.flush()

def loadKeys(mc, keys, data):
    start = time.time()
    random.shuffle(keys)
    for vb, k in keys:
        mc.vbucketId = vb
        mc.set(k, 0, 0, data)
    end = time.time()
    debug("Completed loading in %.2fs (%.2f/s)" % (end - start,
                                                   NUM_KEYS / (end - start)))

def initVBuckets(mc):
    for v in range(VBUCKETS):
        mc.set_vbucket_state(v, 'active')

def runTest(p, name, keys, data):
    while True:
        time.sleep(0.2)
        try:
            mc = mc_bin_client.MemcachedClient()
            initVBuckets(mc)
            break
        except:
            if p.poll():
                debug("Process failed.")
                return False

    start = time.time()
    loadKeys(mc, keys, data)
    stats = mc.stats()
    while int(stats['ep_total_persisted']) != NUM_KEYS:
        time.sleep(0.2)
        stats = mc.stats()
    end = time.time()
    debug("Completed insert on disk in %.2fs" % (end - start))

    partone = ("%s,%s,%s" % (name,
                             stats['ep_commit_time_total'],
                             stats['ep_flush_duration_total']))

    for i in range(UP_ITERS):
        start2 = time.time()
        loadKeys(mc, keys, data)
        stats = mc.stats()
        while int(stats['ep_flusher_todo']) + int(stats['ep_queue_size']):
            time.sleep(0.2)
            stats = mc.stats()
        end = time.time()
        debug("Completed update on disk in %.2fs (+%.2f)" % (end - start,
                                                             end - start2))

    sys.stdout.write("%s,%s,%s\n" % (partone,
                                     stats['ep_commit_time_total'],
                                     stats['ep_flush_duration_total']))
    sys.stdout.flush()
    return True

def doStrategy(strategy, keys, data):
    for f in glob.glob("/mnt/mytest/perftest.db*"):
        os.unlink(f)
    time.sleep(1) # Give it time to settle down
    p = subprocess.Popen(['../memcached/memcached', # '-v',
                          '-E', '.libs/ep.so',
                          '-e', 'dbname=/mnt/mytest/perftest.db;db_strategy=%s;initfile=t/wal.sql' % strategy])
    rv = False
    try:
        rv = runTest(p, strategy, keys, data)
    finally:
        try:
            os.kill(p.pid, 9)
        except:
            sys.stderr.write("Error killing %s" % p.pid)
    return rv

# Let's get a consistent set of data for the given input
random.seed(81748)

start_time = time.time()
keys = set()
while len(keys) < NUM_KEYS:
    keys.add(''.join(random.sample(CHARSET, KEYLEN)))

keys = [(random.randint(0, VBUCKETS-1), k) for k in sorted(keys)]
data = []
while sum(len(x) for x in data) < VALLEN:
    data.append(''.join(random.sample(CHARSET, 32)))
data = ''.join(data)[:VALLEN]
duration = time.time() - start_time
debug("Initialized %d keys and a %d byte value in %.2f seconds" % (
        len(keys), len(data), duration))

strategies = ('multiDB',
              # 'multiMTDB',
              'multiMTVBDB',
             )

for attempt in range(NUM_ATTEMPTS):
    for s in strategies:
        while not doStrategy(s, keys, data):
            debug("Retrying...")