diff options
| author | Serhiy Storchaka <storchaka@gmail.com> | 2015-03-13 06:31:34 (GMT) |
|---|---|---|
| committer | Serhiy Storchaka <storchaka@gmail.com> | 2015-03-13 06:31:34 (GMT) |
| commit | 7c26be5b18d71c1c9863d81f1f478bb803e8bd5c (patch) | |
| tree | 42e71c48092183531bbad9eee645ae1e108aa795 /Lib/multiprocessing | |
| parent | 59bdf6392de446de8a19bfa37cee52981612830e (diff) | |
| download | cpython-7c26be5b18d71c1c9863d81f1f478bb803e8bd5c.zip cpython-7c26be5b18d71c1c9863d81f1f478bb803e8bd5c.tar.gz cpython-7c26be5b18d71c1c9863d81f1f478bb803e8bd5c.tar.bz2 | |
Issue #23051: multiprocessing.Pool methods imap() and imap_unordered() now
handle exceptions raised by an iterator. Patch by Alon Diamant and Davin
Potts.
Diffstat (limited to 'Lib/multiprocessing')
| -rw-r--r-- | Lib/multiprocessing/pool.py | 37 |
1 files changed, 23 insertions, 14 deletions
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 04531b9..991f87f 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -334,25 +334,34 @@ class Pool(object): thread = threading.current_thread() for taskseq, set_length in iter(taskqueue.get, None): + task = None i = -1 - for i, task in enumerate(taskseq): - if thread._state: - debug('task handler found thread._state != RUN') - break - try: - put(task) - except Exception as e: - job, ind = task[:2] + try: + for i, task in enumerate(taskseq): + if thread._state: + debug('task handler found thread._state != RUN') + break try: - cache[job]._set(ind, (False, e)) - except KeyError: - pass - else: + put(task) + except Exception as e: + job, ind = task[:2] + try: + cache[job]._set(ind, (False, e)) + except KeyError: + pass + else: + if set_length: + debug('doing set_length()') + set_length(i+1) + continue + break + except Exception as ex: + job, ind = task[:2] if task else (0, 0) + if job in cache: + cache[job]._set(ind + 1, (False, ex)) if set_length: debug('doing set_length()') set_length(i+1) - continue - break else: debug('task handler got sentinel') |