deadlock using subprocess, pty, and threadpools

Viewed 359

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?

1 Answers

The file descriptor opened as w by one thread may be reused by another thread between the call to os.close in cmd_output_p and the one in pty. In that case, it gets closed unexpectedly by the second close (which would otherwise fail “harmlessly” with EBADF). (In almost any multithreaded program, EBADF must be treated as an assertion failure.)

That certainly can cause the EBADF from the other thread. If the file descriptor gets reopened, it could perhaps cause the deadlock as well since both ends of the pseudoterminal are read-write.

Related