using conncurrent.futures.ProcessPool leads to an OSError("handle is closed") on exiting
When using concurrent.future.ProcessPool
an OSError("handle is closed")
will be raised when the program exists.
The behavior can be seen when running the example from the Python-documenataion:
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 main():
with concurrent.futures.ProcessPoolExecutor() as executor:
for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
print('%d is prime: %s' % (number, prime))
if __name__ == '__main__':
main()`
Running this example with pypy3 version 7.3.2 results in the following Error, after the program has finished.
Traceback (most recent call last): File "/opt/pypy3/lib-python/3/concurrent/futures/process.py", line 102, in _python_exit thread_wakeup.wakeup() File "/opt/pypy3/lib-python/3/concurrent/futures/process.py", line 90, in wakeup self._writer.send_bytes(b"") File "/opt/pypy3/lib-python/3/multiprocessing/connection.py", line 183, in send_bytes self._check_closed() File "/opt/pypy3/lib-python/3/multiprocessing/connection.py", line 136, in _check_closed raise OSError("handle is closed")
It seems that this is a bug in the Python standard library, in particularly in class _ThreadWakeup
in `concurrent/futures/process.py, which has been corrected only in Python 3.9. The Python 3.9 version reads:
`class _ThreadWakeup: def init(self): self._closed = False self._reader, self._writer = mp.Pipe(duplex=False)
def close(self):
if not self._closed:
self._closed = True
self._writer.close()
self._reader.close()
def wakeup(self):
if not self._closed:
self._writer.send_bytes(b"")
def clear(self):
if not self._closed:
while self._reader.poll():
self._reader.recv_bytes()`
while earlier versions read:
`class _ThreadWakeup: def init(self): self._reader, self._writer = mp.Pipe(duplex=False)
def close(self):
self._writer.close()
self._reader.close()
def wakeup(self):
self._writer.send_bytes(b"")
def clear(self):
while self._reader.poll():
self._reader.recv_bytes()`
Updating the definition of class _ThreadWakeup
in lib-python/3/concurrent/futures/process.py
makes the Error go away
process_pool_doc_example.py