threadpool-error-handling-comparison.py

#!/usr/bin/env python

from __future__ import absolute_import, print_function

from multiprocessing.pool import ThreadPool
from random import random
from time import sleep
from timeit import default_timer
import sys


def dodgy_f(job_id):
    rand = random()
    if rand < 0.05:
        raise RuntimeError('Fluctuations in the subspace capacitor')
    else:
        sleep(0.01 * rand)
    return job_id, True


def test_apply_async(pool, jobs):
    print('apply_async')
    total = success = 0

    for result in (pool.apply_async(dodgy_f, (i, )) for i in jobs):
        total += 1

        try:
            result.get()
            if result.successful():
                success += 1
            else:
                print('FAIL')
        except Exception as exc:
            print('Got exception: %s' % exc, file=sys.stderr)

    return total, success


def test_map(pool, jobs):
    print('map')
    total = success = 0

    try:
        for result in pool.map(dodgy_f, jobs):
            total += 1
            success += 1
    except Exception as exc:
        print('Got exception: %s' % exc, file=sys.stderr)

    return total, success


def test_imap(pool, jobs):
    print('imap')
    total = success = 0

    try:
        for result in pool.imap(dodgy_f, jobs):
            total += 1
            success += 1
    except Exception as exc:
        print('Got exception: %s' % exc, file=sys.stderr)

    return total, success


def test_imap_unordered(pool, jobs):
    print('imap_unordered')
    total = success = 0

    try:
        for result in pool.imap_unordered(dodgy_f, jobs):
            total += 1
            success += 1
    except Exception as exc:
        print('Got exception: %s' % exc, file=sys.stderr)

    return total, success


for f in (test_apply_async, test_map, test_imap, test_imap_unordered):
    jobs = range(0, 1000)
    pool = ThreadPool(100)

    start_time = default_timer()

    try:
        total, success = f(pool, jobs)
    except Exception as exc:
        print('%s had an unhandled exception: %s' % (f, exc))
        continue

    elapsed = default_timer() - start_time

    print('%s completed in %0.0fs: total: %4d; Success: %4d' % (f, elapsed, total, success))