summaryrefslogtreecommitdiffstats
path: root/Lib/multiprocessing/pool.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/multiprocessing/pool.py')
-rw-r--r--Lib/multiprocessing/pool.py37
1 files changed, 23 insertions, 14 deletions
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py
index 75a76a4..6d25469 100644
--- a/Lib/multiprocessing/pool.py
+++ b/Lib/multiprocessing/pool.py
@@ -374,25 +374,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:
- util.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:
+ util.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:
+ util.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:
util.debug('doing set_length()')
set_length(i+1)
- continue
- break
else:
util.debug('task handler got sentinel')