| msg326226 - (view) |
Author: Joni Kähärä (kahara) * |
Date: 2018-09-24 12:32 |
| https://docs.python.org/3.8/library/concurrent.futures.html "initializer is an optional callable that is called at the start of each worker process; initargs is a tuple of arguments passed to the initializer. Should initializer raise an exception, all currently pending jobs will raise a BrokenThreadPool, as well any attempt to submit more jobs to the pool." This should be BrokenProcessPool? |
|
|
| msg326231 - (view) |
Author: Karthikeyan Singaravelan (xtreak) *  |
Date: 2018-09-24 12:51 |
| I think it raises BrokenThreadPool . A sample program that I tried as below raising an exception in the initializer. Maybe I am wrong here. Can you please attach a script that triggers BrokenProcessPool? # bpo34786.py import concurrent.futures import time def bar(i): raise Exception(i) # Raise exception from the initializer def foo(i): time.sleep(i) return "1" with concurrent.futures.ThreadPoolExecutor(max_workers=5, initializer=bar, initargs=(1,)) as executor: future_to_url = {executor.submit(foo, i, 60): i for i in range(10)} for future in concurrent.futures.as_completed(future_to_url): try: data = future.result() except Exception as exc: print('generated an exception: %s' % (exc)) else: print('%d bytes' % (len(data))) # Run the program ./python.exe ../backups/bpo34786.py Exception in initializer: Traceback (most recent call last): File "/Users/karthikeyansingaravelan/stuff/python/cpython/Lib/concurrent/futures/thread.py", line 69, in _worker initializer(*initargs) File "../backups/bpo34786.py", line 5, in bar raise Exception(i) Exception: 1 Exception in initializer: Traceback (most recent call last): File "/Users/karthikeyansingaravelan/stuff/python/cpython/Lib/concurrent/futures/thread.py", line 69, in _worker initializer(*initargs) File "../backups/bpo34786.py", line 5, in bar raise Exception(i) Exception: 1 Traceback (most recent call last): File "../backups/bpo34786.py", line 13, in future_to_url = {executor.submit(foo, i, 60): i for i in range(10)} File "../backups/bpo34786.py", line 13, in future_to_url = {executor.submit(foo, i, 60): i for i in range(10)} File "/Users/karthikeyansingaravelan/stuff/python/cpython/Lib/concurrent/futures/thread.py", line 148, in submit raise BrokenThreadPool(self._broken) concurrent.futures.thread.BrokenThreadPool: A thread initializer failed, the thread pool is not usable anymore Thanks |
|
|
| msg326234 - (view) |
Author: Joni Kähärä (kahara) * |
Date: 2018-09-24 13:24 |
| Perhaps I wasn't clear that this considered ProcessPoolExecutor documentation, not ThreadPoolExecutor. Taking the example from documentation and adding an initializer like this: import concurrent.futures import math PRIMES = [ 112272535095293, 112582705942171, 112272535095293, 115280095190773, 115797848077099, 1099726899285419] def is_prime(n): if n % 2 == 0: return False sqrt_n = int(math.floor(math.sqrt(n))) for i in range(3, sqrt_n + 1, 2): if n % i == 0: return False return True def init(): raise Exception() def main(): with concurrent.futures.ProcessPoolExecutor(initializer=init) as executor: for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)): print('%d is prime: %s' % (number, prime)) if __name__ == '__main__': main() ...yields the following: Exception in initializer: Traceback (most recent call last): File "/Users/developer/cpython/Lib/concurrent/futures/process.py", line 219, in _process_worker initializer(*initargs) File "/tmp/bpo34786.py", line 25, in init raise Exception() Exception Exception in initializer: Traceback (most recent call last): File "/Users/developer/cpython/Lib/concurrent/futures/process.py", line 219, in _process_worker initializer(*initargs) File "/tmp/bpo34786.py", line 25, in init raise Exception() Exception Exception in initializer: Traceback (most recent call last): File "/Users/developer/cpython/Lib/concurrent/futures/process.py", line 219, in _process_worker initializer(*initargs) File "/tmp/bpo34786.py", line 25, in init raise Exception() Exception Exception in initializer: Traceback (most recent call last): File "/Users/developer/cpython/Lib/concurrent/futures/process.py", line 219, in _process_worker initializer(*initargs) File "/tmp/bpo34786.py", line 25, in init raise Exception() Exception Traceback (most recent call last): File "/tmp/bpo34786.py", line 33, in main() File "/tmp/bpo34786.py", line 29, in main for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)): File "/Users/developer/cpython/Lib/concurrent/futures/process.py", line 476, in _chain_from_iterable_of_lists for element in iterable: File "/Users/developer/cpython/Lib/concurrent/futures/_base.py", line 594, in result_iterator yield fs.pop().result() File "/Users/developer/cpython/Lib/concurrent/futures/_base.py", line 436, in result return self.__get_result() File "/Users/developer/cpython/Lib/concurrent/futures/_base.py", line 388, in __get_result raise self._exception concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending. |
|
|
| msg326235 - (view) |
Author: Karthikeyan Singaravelan (xtreak) * |
Date: 2018-09-24 13:40 |
| Got it. Sorry about that I copy pasted the quoted string and assumed ThreadPoolExecutor overlooking the PR. The change is reasonable to me. Thanks |
|
|