parallel_file_processing_example.py

import multiprocessing
import argparse
import random
import time

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('-p', '--parallelism', type=parse_int_range(1, 100), default=1,
        help='the number of files to process in parallel (hint: the number of CPUs you want to use)')
    parser.add_argument('paths', metavar='path', nargs='+', help='files to process')

    args = parser.parse_args()

    if args.parallelism == 1:
        # If we only want one process, no need to spawn a bunch of child processes.  Just
        # do the processing directly.
        for path in args.paths:
            process_file(path)
    elif args.parallelism > 1:
        # Create a pool with a fixed number of child processes.
        pool = multiprocessing.Pool(processes=args.parallelism)
        # Distribute the files to the child processes.
        result_handles = []
        for path in args.paths:
            result_handle = pool.apply_async(process_file, (path,))
            result_handles.append(result_handle)
        # Wait for all the tasks to complete.
        for result_handle in result_handles:
            result_handle.get()
    else:
        raise AssertionError("bad value for --parallelism: {!r}".format(args.parallelism))

def parse_int_range(min_value, max_value):
    def f(s):
        try:
            i = int(s)
        except ValueError:
            raise argparse.ArgumentTypeError("expecting an integer, got {!r}".format(s))
        if i < min_value or i > max_value:
            raise argparse.ArgumentTypeError("expecting an integer from {} to {}, got {}".format(min_value, max_value, i))
        return i
    return f

def process_file(path):
    print("start {!r}".format(path))
    # TODO: Actually process file instead of just sleeping.
    time.sleep(2 + random.random() * 2)
    print("finish {!r}".format(path))

if __name__ == '__main__':
    main()