diff -r 4d8e37e54a7d Lib/multiprocessing/pool.py --- a/Lib/multiprocessing/pool.py Sat Feb 21 19:51:17 2015 +0200 +++ b/Lib/multiprocessing/pool.py Mon Feb 23 01:54:37 2015 -0600 @@ -334,25 +334,34 @@ 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') diff -r 4d8e37e54a7d Lib/test/test_multiprocessing.py --- a/Lib/test/test_multiprocessing.py Sat Feb 21 19:51:17 2015 +0200 +++ b/Lib/test/test_multiprocessing.py Mon Feb 23 01:54:37 2015 -0600 @@ -1102,6 +1102,13 @@ def sqr(x, wait=0.0): time.sleep(wait) return x*x + +def exception_throwing_generator(total, when): + for i in range(total): + if i == when: + raise ValueError("Somebody said when") + yield i + class _TestPool(BaseTestCase): def test_apply(self): @@ -1157,6 +1164,25 @@ self.assertEqual(it.next(), i*i) self.assertRaises(StopIteration, it.next) + def test_imap_handle_iterable_exception(self): + if self.TYPE == 'manager': + self.skipTest('test not appropriate for {}'.format(self.TYPE)) + + it = self.pool.imap(sqr, exception_throwing_generator(10, 3), 1) + for i in range(3): + self.assertEqual(next(it), i*i) + self.assertRaises(ValueError, it.next) + + # ValueError seen at start of problematic chunk's results + it = self.pool.imap(sqr, exception_throwing_generator(20, 7), 2) + for i in range(6): + self.assertEqual(next(it), i*i) + self.assertRaises(ValueError, it.next) + it = self.pool.imap(sqr, exception_throwing_generator(20, 7), 4) + for i in range(4): + self.assertEqual(next(it), i*i) + self.assertRaises(ValueError, it.next) + def test_imap_unordered(self): it = self.pool.imap_unordered(sqr, range(1000)) self.assertEqual(sorted(it), map(sqr, range(1000))) @@ -1164,6 +1190,25 @@ it = self.pool.imap_unordered(sqr, range(1000), chunksize=53) self.assertEqual(sorted(it), map(sqr, range(1000))) + def test_imap_unordered_handle_iterable_exception(self): + if self.TYPE == 'manager': + self.skipTest('test not appropriate for {}'.format(self.TYPE)) + + it = self.pool.imap_unordered(sqr, + exception_throwing_generator(10, 3), + 1) + with self.assertRaises(ValueError): + # imap_unordered makes it difficult to anticipate the ValueError + for i in range(10): + self.assertEqual(next(it), i*i) + + it = self.pool.imap_unordered(sqr, + exception_throwing_generator(20, 7), + 2) + with self.assertRaises(ValueError): + for i in range(20): + self.assertEqual(next(it), i*i) + def test_make_pool(self): self.assertRaises(ValueError, multiprocessing.Pool, -1) self.assertRaises(ValueError, multiprocessing.Pool, 0)