Create a gist now

Instantly share code, notes, and snippets.

What would you like to do?
test_codis_mem.py
import threading
import os
import time
import commands
import redis
class RedisServer(threading.Thread):
def __init__(self, binary = 'redis-server', port = 7777):
threading.Thread.__init__(self)
self.binary = binary
self.port = port
def run(self):
cmd = '%s --port %s' % (self.binary, self.port)
print cmd
#os.system(cmd)
commands.getoutput(cmd)
def __str__(self):
return '%s:%s' % (self.binary, self.port)
r1 = RedisServer('./bin/codis-server', port = 7777)
r2 = RedisServer('./bin/redis-server', port = 7778)
def test_mem(r):
r.start()
time.sleep(1)
conn = redis.StrictRedis('127.0.0.1', r.port)
ret = {}
ret[0] = conn.info()['used_memory']
i = 0
while i < 1000:
conn.set('k-%s' % i, 'x')
i += 1
ret[1000] = conn.info()['used_memory']
while i < 10000:
conn.set('k-%s' % i, 'x')
i += 1
ret[10000] = conn.info()['used_memory']
while i < 100000:
conn.set('k-%s' % i, 'x')
i += 1
ret[100000] = conn.info()['used_memory']
while i < 1000000:
conn.set('k-%s' % i, 'x')
i += 1
ret[1000000] = conn.info()['used_memory']
print r, ret
test_mem(r1)
test_mem(r2)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment