使用子进程，pty和线程池的死锁

我有，我想假一TTY连接至在一个子进程运行在特定情况下ThreadPoolExecutor（这样想xargs -p），并捕获输出。

我创建了以下内容，这些内容似乎可以连续工作：

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

和一些示例用法：

>>> cmd_output_p('python', '-c', 'import sys; print(sys.stdout.isatty())')
(0, b'True\r\n')

太棒了！但是，当我在相同的过程中运行时，concurrent.futures.ThreadPoolExecutor会有很多故障模式（还有另外一种较为罕见的故障模式，即OSError: [Errno 9] Bad file descriptor在看似随机的代码行上发生的代码-但我还没有隔离该代码的复制品）

例如以下代码：

def run(arg):
    print(cmd_output_p('echo', arg))


with concurrent.futures.ThreadPoolExecutor(5) as exe:
    exe.map(run, ('1',) * 1000)

这会经过一系列过程，然后最终挂起。发出^C以下堆栈跟踪（需要两次^ C才能结束进程）

$ 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

大概我做错了什么，但是我发现与子流程/ pty交互的所有示例都执行相同的步骤-如何防止这种死锁？