Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
#!/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))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.