Demonstrate some funny stuffs of memcached.

memcached_demo.py

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

import sys
import time
import math
import re
import telnetlib
import random

from collections import namedtuple, defaultdict

import pylibmc


class MemcachedStats:
    """Dump memcached stats data"""
    _stats_regex = re.compile(ur"STAT (\w+) (.*)\r")
    _items_regex = re.compile(ur'STAT items:(.*):(.*) (\d+)\r')
    _slabs_regex = re.compile(ur'STAT (\d+):(.*) (\d+)\r')

    def __init__(self, host='localhost', port='11211'):
        self._client = telnetlib.Telnet(host, port)

    def _cast(self, val):
        try:
            return int(val)
        except ValueError:
            return val

    def _command(self, cmd):
        self._client.write('%s\n' % cmd)
        return self._client.read_until('END')

    @property
    def settings(self):
        raw = zip(*self._stats_regex.findall(self._command('stats settings')))
        return dict(zip(raw[0], map(self._cast, raw[1])))

    @property
    def items(self):
        data = defaultdict(dict)
        for id, key, val in self._items_regex.findall(self._command('stats items')):
            data[id][key] = self._cast(val)
        return dict(data)

    @property
    def slabs(self):
        msg = self._command('stats slabs')
        data = { 'slabs': defaultdict(dict) }
        for key, val in self._stats_regex.findall(msg):
            data[key] = self._cast(val)
        for id, key, val in self._slabs_regex.findall(msg):
            data['slabs'][id][key] = self._cast(val)
        data['slabs'] = dict(data['slabs'])
        return data

    @property
    def sizes(self):
        raw = zip(*self._stats_regex.findall(self._command('stats sizes')))
        return dict(zip(raw[0], map(self._cast, raw[1])))

    @property
    def info(self):
        raw = zip(*self._stats_regex.findall(self._command('stats')))
        return dict(zip(raw[0], map(self._cast, raw[1])))


_random = random.SystemRandom()
def random_string(length=12, chars='abcdefghijklmnopqrstuvwxyz'
        'ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789'):
    return ''.join([_random.choice(chars) for i in range(length)])


def populate(client, n, key_len=16, data_len=128, expire=0):
    bar = random_string(data_len)
    for i in xrange(n):
        client.set(random_string(key_len), bar, time=expire)


def calculate_size(key_size, data_size):
    size = sum((
            48,                     # Size of item structure in memcached
            key_size + 1,           # Key size
            min(40, len(' %d %d\r\n' % (0, data_size))),        # Suffix defined in memcached. Assume flags is 0
            data_size + 2,          # Data size with addition CRLF terminator
            8,                      # Use CAS. Size of uint64_t
        ))
    return size


def test_storage(client, settings, stats):
    Test = namedtuple('Test', 'item_num, key_len, data_len')
    tests = [
        Test(3000, 8, 8),                # 8B
        Test(1000, 8, 128),              # 128B
        Test(500, 16, 1024),             # 1K
        Test(20, 32, 1024 * 256),       # 256K
    ]

    output_tpl = """>>>> Test #%(index)d
    Item Number:        %(item_num)d
    Key Size:           %(key_size)d
    Data Size:          %(data_size)d
    Item Size:          %(item_size)d
    Total Memory:       %(total_size)d

    Slab Index:         %(slab_index)d
    Chunk Size:         %(chunk_size)d
    Total Pages:        %(total_pages)d
    Total Chunks:       %(total_chunks)d
    Used Chunks:        %(used_chunks)d
    Requested Memory:   %(mem_requested)d
    Used Memory:        %(used_memory)d
    Wasted:             %(wasted_memory)d
"""

    for idx, t in enumerate(tests, 1):
        key_size = t.key_len
        data_size = t.data_len
        item_size = calculate_size(key_size, data_size)
        total_size = item_size * t.item_num

        slab_index = 0
        while (settings['slab_classes'][slab_index] < item_size):
            slab_index += 1
        slab_index += 1         # 1-based

        populate(client, t.item_num, t.key_len, t.data_len)

        slab_stats = stats.slabs['slabs'][str(slab_index)]
        data = {
            'index': idx,
            'item_num': t.item_num,
            'key_size': key_size,
            'data_size': data_size,
            'item_size': item_size,
            'total_size': total_size,
            'slab_index': slab_index,
            'chunk_size': slab_stats['chunk_size'],
            'total_pages': slab_stats['total_pages'],
            'total_chunks': slab_stats['total_chunks'],
            'used_chunks': slab_stats['used_chunks'],
            'mem_requested': slab_stats['mem_requested'],
            'used_memory': slab_stats['chunk_size'] * slab_stats['used_chunks'],
            'wasted_memory': slab_stats['chunk_size'] * slab_stats['used_chunks'] - slab_stats['mem_requested'],
        }

        print output_tpl % data


def test_expire(client, settings, stats):
    def print_condition(item_size, n):
        print '    Item size: %d' % item_size
        print '    Item number: %d' % n
        print '    Total size: %d' % (item_size * n)

    def print_stats(slabs):
        for slab_index in slabs:
            slab_stats = stats.slabs['slabs'][str(slab_index)]
            item_stats = stats.items[str(slab_index)]
            print '    Slab #%d' % slab_index
            print '      Requested memory: %d' % slab_stats['mem_requested']
            print '      Used memory: %d' % (slab_stats['chunk_size'] * slab_stats['used_chunks'])
            print '      Evicted: %d' % item_stats['evicted']

    print '>>>> 1st Population - Use 2/3 Memory and expire in 10s'
    key_len, data_len, expire = 32, 1024, 10
    item_size = calculate_size(key_len, data_len)
    n = int(settings['max_memory'] * 1024 * 1024 / 3 * 2 / item_size)
    print_condition(item_size, n)

    populate(client, n, key_len, data_len, expire)
    print_stats([12])       # I just know it's slab 12...

    sys.stdout.write('\nSleep 12s waiting data expired...')
    sys.stdout.flush()
    for i in xrange(12):
        time.sleep(1)
        sys.stdout.write('.')
        sys.stdout.flush()
    sys.stdout.write('\n\n')

    print '>>>> 2nd Population - Use 2/3 Memory with a different slab size'
    key_len, data_len, expire = 32, 256, 0
    item_size = calculate_size(key_len, data_len)
    n = int(settings['max_memory'] * 1024 * 1024 / 3 * 2 / item_size)
    print_condition(item_size, n)

    populate(client, n, key_len, data_len, expire)
    print_stats([7, 12])

    sys.stdout.write('\n')

    print '>>>> 3rd Population - Use 2/3 Memory. Same size of 1st Population and never expire'
    key_len, data_len, expire = 32, 1024, 0
    item_size = calculate_size(key_len, data_len)
    n = int(settings['max_memory'] * 1024 * 1024 / 3 * 2 / item_size)
    print_condition(item_size, n)

    populate(client, n, key_len, data_len, expire)
    print_stats([7, 12])

    sys.stdout.write('\n')

    print '>>>> 4th Population - Use 2/3 Memory. Same as 3rd Population'
    key_len, data_len, expire = 32, 1024, 0
    item_size = calculate_size(key_len, data_len)
    n = int(settings['max_memory'] * 1024 * 1024 / 3 * 2 / item_size)
    print_condition(item_size, n)

    populate(client, n, key_len, data_len, expire)
    print_stats([7, 12])


if __name__ == '__main__':
    # Memcached configuration: memcached -m 16 -n 48 -f 1.25 -I 1048576
    mc_settings = {
        'max_memory': 16,
        'min_space': 48,
        'growth_factor': 1.25,
        'item_max_size': 1048576,      # 1m. dEFAULT SIZE OF EACH SLAB PAGE
    }

    # Get memcached slab classes
    slab_classes = [mc_settings['min_space'] + 48, ]        # First slab size plus sizeof(item)
    while slab_classes[-1] < mc_settings['item_max_size'] / mc_settings['growth_factor']:
        size = slab_classes[-1] * mc_settings['growth_factor']
        if size % 8:         # Always 8-bytes aligned
            size += 8 - (size % 8)
        slab_classes.append(int(size))
    slab_classes[-1] = mc_settings['item_max_size']
    mc_settings['slab_classes'] = slab_classes

    mc_client = pylibmc.Client(['127.0.0.1:11211'])
    mc_stats = MemcachedStats()
    test_storage(mc_client, mc_settings, mc_stats)

    print 'Restart memcached to run next test'
    raw_input()

    mc_client = pylibmc.Client(['127.0.0.1:11211'])
    mc_stats = MemcachedStats()
    test_expire(mc_client, mc_settings, mc_stats)