concurrent map

concurrent_map.py

## {{{ http://code.activestate.com/recipes/577360/ (r1)
import threading

def concurrent_map(func, data):
    """
    Similar to the bultin function map(). But spawn a thread for each argument
    and apply `func` concurrently.

    Note: unlike map(), we cannot take an iterable argument. `data` should be an
    indexable sequence.
    """

    N = len(data)
    result = [None] * N

    # wrapper to dispose the result in the right slot
    def task_wrapper(i):
        result[i] = func(data[i])

    threads = [threading.Thread(target=task_wrapper, args=(i,)) for i in xrange(N)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    return result
## end of http://code.activestate.com/recipes/577360/ }}}

dmap1.py

def dmap1(func, data):
    """distributed map (v1) derived from concurrent_map with supporting of iterable argument.
    """

    result = {}

    # wrapper to dispose the result in the right slot
    def task_wrapper(i, d):
        result[i] = func(d)

    threads = [threading.Thread(target=task_wrapper, args=(i, d)) for (i, d) in enumerate(data)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    return [result[i] for i in xrange(len(result))]

dmap2.py

def dmap2(func, *data):
    """distributed map (v2) derived from dmap1 with supporting of multiple arguments.
    """

    result = {}

    # wrapper to dispose the result in the right slot
    def task_wrapper(i, t):
        result[i] = func(*list(t))

    threads = [threading.Thread(target=task_wrapper, args=(i, t)) for (i, t) in enumerate(zip(*data))]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    return [result[i] for i in xrange(len(result))]

blog: http://drakeguan.org/node/639