#

# Simple benchmarks for the multiprocessing package

#

import time, sys, multiprocessing, threading, Queue, gc

if sys.platform == 'win32':

    _timer = time.clock

else:

    _timer = time.time

delta = 1

#### TEST_QUEUESPEED

def queuespeed_func(q, c, iterations):

    a = '0' * 256

    c.acquire()

    c.notify()

    c.release()

    for i in xrange(iterations):

        q.put(a)

    q.put('STOP')

def test_queuespeed(Process, q, c):

    elapsed = 0

    iterations = 1

    while elapsed < delta:

        iterations *= 2

        p = Process(target=queuespeed_func, args=(q, c, iterations))

        c.acquire()

        p.start()

        c.wait()

        c.release()

        result = None

        t = _timer()

        while result != 'STOP':

            result = q.get()

        elapsed = _timer() - t

        p.join()

    print iterations, 'objects passed through the queue in', elapsed, 'seconds'

    print 'average number/sec:', iterations/elapsed

#### TEST_PIPESPEED

def pipe_func(c, cond, iterations):

    a = '0' * 256

    cond.acquire()

    cond.notify()

    cond.release()

    for i in xrange(iterations):

        c.send(a)

    c.send('STOP')

def test_pipespeed():

    c, d = multiprocessing.Pipe()

    cond = multiprocessing.Condition()

    elapsed = 0

    iterations = 1

    while elapsed < delta:

        iterations *= 2

        p = multiprocessing.Process(target=pipe_func,

                                    args=(d, cond, iterations))

        cond.acquire()

        p.start()

        cond.wait()

        cond.release()

        result = None

        t = _timer()

        while result != 'STOP':

            result = c.recv()

        elapsed = _timer() - t

        p.join()

    print iterations, 'objects passed through connection in',elapsed,'seconds'

    print 'average number/sec:', iterations/elapsed

#### TEST_SEQSPEED

def test_seqspeed(seq):

    elapsed = 0

    iterations = 1

    while elapsed < delta:

        iterations *= 2

        t = _timer()

        for i in xrange(iterations):

            a = seq[5]

        elapsed = _timer()-t

    print iterations, 'iterations in', elapsed, 'seconds'

    print 'average number/sec:', iterations/elapsed

#### TEST_LOCK

def test_lockspeed(l):

    elapsed = 0

    iterations = 1

    while elapsed < delta:

        iterations *= 2

        t = _timer()

        for i in xrange(iterations):

            l.acquire()

            l.release()

        elapsed = _timer()-t

    print iterations, 'iterations in', elapsed, 'seconds'

    print 'average number/sec:', iterations/elapsed

#### TEST_CONDITION

def conditionspeed_func(c, N):

    c.acquire()

    c.notify()

    for i in xrange(N):

        c.wait()

        c.notify()

    c.release()

def test_conditionspeed(Process, c):

    elapsed = 0

    iterations = 1

    while elapsed < delta:

        iterations *= 2

        c.acquire()

        p = Process(target=conditionspeed_func, args=(c, iterations))

        p.start()

        c.wait()

        t = _timer()

        for i in xrange(iterations):

            c.notify()

            c.wait()

        elapsed = _timer()-t

        c.release()

        p.join()

    print iterations * 2, 'waits in', elapsed, 'seconds'

    print 'average number/sec:', iterations * 2 / elapsed

####

def test():

    manager = multiprocessing.Manager()

    gc.disable()

    print '\n\t######## testing Queue.Queue\n'

    test_queuespeed(threading.Thread, Queue.Queue(),

                    threading.Condition())

    print '\n\t######## testing multiprocessing.Queue\n'

    test_queuespeed(multiprocessing.Process, multiprocessing.Queue(),

                    multiprocessing.Condition())

    print '\n\t######## testing Queue managed by server process\n'

    test_queuespeed(multiprocessing.Process, manager.Queue(),

                    manager.Condition())

    print '\n\t######## testing multiprocessing.Pipe\n'

    test_pipespeed()

    print

    print '\n\t######## testing list\n'

    test_seqspeed(range(10))

    print '\n\t######## testing list managed by server process\n'

    test_seqspeed(manager.list(range(10)))

    print '\n\t######## testing Array("i", ..., lock=False)\n'

    test_seqspeed(multiprocessing.Array('i', range(10), lock=False))

    print '\n\t######## testing Array("i", ..., lock=True)\n'

    test_seqspeed(multiprocessing.Array('i', range(10), lock=True))

    print

    print '\n\t######## testing threading.Lock\n'

    test_lockspeed(threading.Lock())

    print '\n\t######## testing threading.RLock\n'

    test_lockspeed(threading.RLock())

    print '\n\t######## testing multiprocessing.Lock\n'

    test_lockspeed(multiprocessing.Lock())

    print '\n\t######## testing multiprocessing.RLock\n'

    test_lockspeed(multiprocessing.RLock())

    print '\n\t######## testing lock managed by server process\n'

    test_lockspeed(manager.Lock())

    print '\n\t######## testing rlock managed by server process\n'

    test_lockspeed(manager.RLock())

    print

    print '\n\t######## testing threading.Condition\n'

    test_conditionspeed(threading.Thread, threading.Condition())

    print '\n\t######## testing multiprocessing.Condition\n'

    test_conditionspeed(multiprocessing.Process, multiprocessing.Condition())

    print '\n\t######## testing condition managed by a server process\n'

    test_conditionspeed(multiprocessing.Process, manager.Condition())

    gc.enable()

if __name__ == '__main__':

    multiprocessing.freeze_support()

    test()