compres-test.txt

Test (in-process): 46044 [ms]
Test (single thread): 44425 [ms]
Test (2 threads): 27837 [ms]
Test (4 threads): 49647 [ms]
Test (8 threads): 43913 [ms]
Test (16 threads): 56567 [ms]
Test (single proc): 49173 [ms]
Test (2 procs): 25880 [ms]
Test (4 procs): 13424 [ms]
Test (8 procs): 11163 [ms]
Test (16 procs): 11333 [ms]

decompress-test.txt

$ python thread-test.py 
Test (in-process): 8745 [ms]
Test (single thread): 8766 [ms]
Test (2 threads): 4604 [ms]
Test (4 threads): 2826 [ms]
Test (8 threads): 6241 [ms]
Test (16 threads): 5831 [ms]
Test (single proc): 8836 [ms]
Test (2 procs): 4480 [ms]
Test (4 procs): 2351 [ms]
Test (8 procs): 1731 [ms]
Test (16 procs): 1743 [ms]

zlib_thread.py

#!/usr/bin/env python

import threading
import multiprocessing
import random
import string

from zlib import compress as zlib_compress, decompress as zlib_decompress
from time import time

def build_test_func(compress=True, iterations=10000, data=8 * 1024, level=9):
    try:
        data = ''.join(random.choice(string.letters) for _ in xrange(data))
    except ValueError:
        pass
    if compress:
        work_func = lambda d: zlib_compress(d, level)
    else:
        data = zlib_compress(data, level)
        work_func = lambda d: zlib_decompress(d)

    def test_func():
        for _ in xrange(iterations):
            work_func(data)
    return test_func

def run_threads(count, worker):
    threads = [threading.Thread(target=worker) for _ in range(count)]
    map(lambda t: t.start(), threads)
    map(lambda t: t.join(), threads)

def run_procs(count, worker):
    procs = [multiprocessing.Process(target=worker) for _ in range(count)]
    map(lambda p: p.start(), procs)
    map(lambda p: p.join(), procs)

def run_test(test_func, test_name, cycles=1):
    loop = range(cycles)
    t = time()
    for _ in loop:
        test_func()
    t = time() - t
    print 'Test (%s): %d [ms]' % (test_name, t * 1000)

test_func = build_test_func(compress=False, iterations=6*10**4, data=0x800, level=9)

run_test(test_func, 'in-process', 16)
run_test(lambda: run_threads(1, test_func), 'single thread', 16)
run_test(lambda: run_threads(2, test_func), '2 threads', 8)
run_test(lambda: run_threads(4, test_func), '4 threads', 4)
run_test(lambda: run_threads(8, test_func), '8 threads', 2)
run_test(lambda: run_threads(16, test_func), '16 threads')
run_test(lambda: run_procs(1, test_func), 'single proc', 16)
run_test(lambda: run_procs(2, test_func), '2 procs', 8)
run_test(lambda: run_procs(4, test_func), '4 procs', 4)
run_test(lambda: run_procs(8, test_func), '8 procs', 2)
run_test(lambda: run_procs(16, test_func), '16 procs')