diff options
| -rw-r--r-- | Lib/multiprocessing/pool.py | 14 | ||||
| -rw-r--r-- | Lib/test/_test_multiprocessing.py | 10 |
2 files changed, 19 insertions, 5 deletions
diff --git a/Lib/multiprocessing/pool.py b/Lib/multiprocessing/pool.py index 4be00d5..1cb2d95 100644 --- a/Lib/multiprocessing/pool.py +++ b/Lib/multiprocessing/pool.py @@ -174,7 +174,8 @@ class Pool(object): self._task_handler = threading.Thread( target=Pool._handle_tasks, - args=(self._taskqueue, self._quick_put, self._outqueue, self._pool) + args=(self._taskqueue, self._quick_put, self._outqueue, + self._pool, self._cache) ) self._task_handler.daemon = True self._task_handler._state = RUN @@ -364,7 +365,7 @@ class Pool(object): util.debug('worker handler exiting') @staticmethod - def _handle_tasks(taskqueue, put, outqueue, pool): + def _handle_tasks(taskqueue, put, outqueue, pool, cache): thread = threading.current_thread() for taskseq, set_length in iter(taskqueue.get, None): @@ -375,9 +376,12 @@ class Pool(object): break try: put(task) - except OSError: - util.debug('could not put task on queue') - break + 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()') diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py index eead142..f12a5f9 100644 --- a/Lib/test/_test_multiprocessing.py +++ b/Lib/test/_test_multiprocessing.py @@ -1698,6 +1698,16 @@ class _TestPool(BaseTestCase): self.assertEqual(2, len(call_args)) self.assertIsInstance(call_args[1], ValueError) + def test_map_unplicklable(self): + # Issue #19425 -- failure to pickle should not cause a hang + if self.TYPE == 'threads': + return + class A(object): + def __reduce__(self): + raise RuntimeError('cannot pickle') + with self.assertRaises(RuntimeError): + self.pool.map(sqr, [A()]*10) + def test_map_chunksize(self): try: self.pool.map_async(sqr, [], chunksize=1).get(timeout=TIMEOUT1) |