Skip to content

Commit

Permalink
bpo-29704: Fix asyncio.SubprocessStreamProtocol closing (#405)
Browse files Browse the repository at this point in the history
  • Loading branch information
SethMichaelLarson authored and 1st1 committed Mar 3, 2017
1 parent 13802a3 commit 604faba
Show file tree
Hide file tree
Showing 3 changed files with 824 additions and 2 deletions.
17 changes: 15 additions & 2 deletions Lib/asyncio/subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@ def __init__(self, limit, loop):
self._limit = limit
self.stdin = self.stdout = self.stderr = None
self._transport = None
self._process_exited = False
self._pipe_fds = []

def __repr__(self):
info = [self.__class__.__name__]
Expand All @@ -43,12 +45,14 @@ def connection_made(self, transport):
self.stdout = streams.StreamReader(limit=self._limit,
loop=self._loop)
self.stdout.set_transport(stdout_transport)
self._pipe_fds.append(1)

stderr_transport = transport.get_pipe_transport(2)
if stderr_transport is not None:
self.stderr = streams.StreamReader(limit=self._limit,
loop=self._loop)
self.stderr.set_transport(stderr_transport)
self._pipe_fds.append(2)

stdin_transport = transport.get_pipe_transport(0)
if stdin_transport is not None:
Expand Down Expand Up @@ -86,9 +90,18 @@ def pipe_connection_lost(self, fd, exc):
else:
reader.set_exception(exc)

if fd in self._pipe_fds:
self._pipe_fds.remove(fd)
self._maybe_close_transport()

def process_exited(self):
self._transport.close()
self._transport = None
self._process_exited = True
self._maybe_close_transport()

def _maybe_close_transport(self):
if len(self._pipe_fds) == 0 and self._process_exited:
self._transport.close()
self._transport = None


class Process:
Expand Down
24 changes: 24 additions & 0 deletions Lib/test/test_asyncio/test_subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -459,6 +459,30 @@ def test_popen_error(self):
self.loop.run_until_complete(create)
self.assertEqual(warns, [])

def test_read_stdout_after_process_exit(self):
@asyncio.coroutine
def execute():
code = '\n'.join(['import sys',
'for _ in range(64):',
' sys.stdout.write("x" * 4096)',
'sys.stdout.flush()',
'sys.exit(1)'])

fut = asyncio.create_subprocess_exec(
sys.executable, '-c', code,
stdout=asyncio.subprocess.PIPE,
loop=self.loop)

process = yield from fut
while True:
data = yield from process.stdout.read(65536)
if data:
yield from asyncio.sleep(0.3, loop=self.loop)
else:
break

self.loop.run_until_complete(execute())


if sys.platform != 'win32':
# Unix
Expand Down
Loading

0 comments on commit 604faba

Please sign in to comment.