I have a particular case where I'd like to fake a tty to subprocesses which are run in a ThreadPoolExecutor (think like xargs -p) and capture the output.
I've created the following which seems to work well serially:
import contextlib
import concurrent.futures
import errno
import os
import subprocess
import termios
@contextlib.contextmanager
def pty():
r, w = os.openpty()
try:
yield r, w
finally:
for fd in r, w:
try:
os.close(fd)
except OSError:
pass
def cmd_output_p(*cmd):
with pty() as (r, w):
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=w, stderr=w)
os.close(w)
proc.stdin.close()
buf = b''
while True:
try:
bts = os.read(r, 1024)
except OSError as e:
if e.errno == errno.EIO:
bts = b''
else:
raise
else:
buf += bts
if not bts:
break
return proc.wait(), buf
and some sample usage:
>>> cmd_output_p('python', '-c', 'import sys; print(sys.stdout.isatty())')
(0, b'True\r\n')
awesome! however, when I run this same procedure in a concurrent.futures.ThreadPoolExecutor there are quite a few failure modes (there's an additional rarer failure mode where OSError: [Errno 9] Bad file descriptor on a seemingly random line of code occurs -- but I haven't isolated a reproduction for that yet)
for example the following code:
def run(arg):
print(cmd_output_p('echo', arg))
with concurrent.futures.ThreadPoolExecutor(5) as exe:
exe.map(run, ('1',) * 1000)
This chugs through a bunch of processes and then eventually hangs. Issuing ^C gives the following stacktrace(s) (^C is needed twice to end the process)
$ python3 t.py
...
(0, b'1\r\n')
(0, b'1\r\n')
(0, b'1\r\n')
^CTraceback (most recent call last):
File "t.py", line 49, in <module>
exe.map(run, ('1',) * 1000)
File "/usr/lib/python3.6/concurrent/futures/_base.py", line 611, in __exit__
self.shutdown(wait=True)
File "/usr/lib/python3.6/concurrent/futures/thread.py", line 152, in shutdown
t.join()
File "/usr/lib/python3.6/threading.py", line 1056, in join
self._wait_for_tstate_lock()
File "/usr/lib/python3.6/threading.py", line 1072, in _wait_for_tstate_lock
elif lock.acquire(block, timeout):
KeyboardInterrupt
^CError in atexit._run_exitfuncs:
Traceback (most recent call last):
File "/usr/lib/python3.6/concurrent/futures/thread.py", line 40, in _python_exit
t.join()
File "/usr/lib/python3.6/threading.py", line 1056, in join
self._wait_for_tstate_lock()
File "/usr/lib/python3.6/threading.py", line 1072, in _wait_for_tstate_lock
elif lock.acquire(block, timeout):
KeyboardInterrupt
presumably I'm doing something wrong but all of the examples I've found for interaction with subprocess / pty perform the same steps -- how do I prevent this deadlock?