<commit_before>"""
Processes and parallelism
=========================

The ``processes`` module provides some convenience functions
for using processes in python.

Adapted from http://stackoverflow.com/a/16071616/287297

Example usage:

    print prll_map(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8], 32, verbose=True)

Comments:

"It spawns a predefined amount of workers and only iterates through the input list
 if there exists an idle worker. I also enabled the "daemon" mode for the workers so
 that KeyboardInterupt works as expected."

Pitfalls: all the stdouts are sent back to the parent stdout, intertwined.

Alternatively, use this fork of multiprocessing: https://github.com/uqfoundation/multiprocess
"""

# Modules #
import multiprocessing
from tqdm import tqdm

################################################################################
def target_func(f, q_in, q_out):
    while True:
        i, x = q_in.get()
        if i is None: break
        q_out.put((i, f(x)))

################################################################################
def prll_map(func_to_apply, items, cpus=None, verbose=False):
    # Number of cores #
    if cpus is None: cpus = min(multiprocessing.cpu_count(), 32)
    # Create queues #
    q_in = multiprocessing.Queue(1)
    q_out = multiprocessing.Queue()
    # Process list #
    new_proc = lambda t,a: multiprocessing.Process(target=t, args=a)
    processes = [new_proc(target_func, (func_to_apply, q_in, q_out)) for x in range(cpus)]
    # Start them all #
    for proc in processes:
        proc.daemon = True
        proc.start()
    # Put and get #
    sent = [q_in.put((i, x)) for i, x in enumerate(items)]
    for x in range(cpus): q_in.put((None, None))
    res  = [q_out.get() for x in range(len(sent))]
    # Wait for them to finish #
    if verbose:
        for proc in tqdm(processes): proc.join()
    else:
        for proc in processes: proc.join()
    # Return results #
    return [x for i, x in sorted(res)]<commit_msg>Better prll_map<commit_after>"""
Processes and parallelism
=========================

The ``processes`` module provides some convenience functions
for using parallel processes in python.

Adapted from http://stackoverflow.com/a/16071616/287297

Example usage:

    print prll_map(lambda i: i * 2, [1, 2, 3, 4, 6, 7, 8], 32, verbose=True)

Comments:

"It spawns a predefined amount of workers and only iterates through the input list
 if there exists an idle worker. I also enabled the "daemon" mode for the workers so
 that KeyboardInterupt works as expected."

Pitfalls: all the stdouts are sent back to the parent stdout, intertwined.

Alternatively, use this fork of multiprocessing:
https://github.com/uqfoundation/multiprocess
"""

# Modules #
import multiprocessing
from tqdm import tqdm

################################################################################
def apply_function(func_to_apply, queue_in, queue_out):
    while not queue_in.empty():
        num, obj = queue_in.get()
        queue_out.put((num, func_to_apply(obj)))

################################################################################
def prll_map(func_to_apply, items, cpus=None, verbose=True):
    # Number of processes to use #
    if cpus is None: cpus = min(multiprocessing.cpu_count(), 32)
    # Create queues #
    q_in  = multiprocessing.Queue()
    q_out = multiprocessing.Queue()
    # Process list #
    new_proc  = lambda t,a: multiprocessing.Process(target=t, args=a)
    processes = [new_proc(apply_function, (func_to_apply, q_in, q_out)) for x in range(cpus)]
    # Put all the items (objects) in the queue #
    sent = [q_in.put((i, x)) for i, x in enumerate(items)]
    # Start them all #
    for proc in processes:
        proc.daemon = True
        proc.start()
    # Display progress bar or not #
    if verbose:
        results = [q_out.get() for x in tqdm(range(len(sent)))]
    else:
        results = [q_out.get() for x in range(len(sent))]
    # Wait for them to finish #
    for proc in processes: proc.join()
    # Return results #
    return [x for i, x in sorted(results)]

################################################################################
def test():
    def slow_square(x):
        import time
        time.sleep(2)
        return x**2
    objs    = range(20)
    squares = prll_map(slow_square, objs, 4, verbose=True)
    print "Result: %s" % squares
{'fraction_non_alphanumeric': 0.19639598540145986, 'fraction_numerical': 0.01322992700729927, 'mean_word_length': 3.8028477546549837, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '620890', 'n_tokens_mistral': 1405, 'n_tokens_neox': 1327, 'n_words': 444}