[Python-checkins] cpython (3.4): Closes #22685, asyncio: Set the transport of stdout and stderr StreamReader

Tue Nov 25 17:25:02 CET 2014

https://hg.python.org/cpython/rev/f75d40c02f0a
changeset: 93582:f75d40c02f0a
branch: 3.4
parent: 93576:268ceaa78cf9
user: Victor Stinner <victor.stinner at gmail.com>
date: Tue Nov 25 17:20:33 2014 +0100
summary:
 Closes #22685, asyncio: Set the transport of stdout and stderr StreamReader
objects in the SubprocessStreamProtocol. It allows to pause the transport to
not buffer too much stdout or stderr data.
files:
 Lib/asyncio/subprocess.py | 17 ++++-
 Lib/test/test_asyncio/test_subprocess.py | 32 ++++++++++++
 2 files changed, 44 insertions(+), 5 deletions(-)

diff --git a/Lib/asyncio/subprocess.py b/Lib/asyncio/subprocess.py
--- a/Lib/asyncio/subprocess.py
+++ b/Lib/asyncio/subprocess.py
@@ -41,15 +41,22 @@
 
 def connection_made(self, transport):
 self._transport = transport
- if transport.get_pipe_transport(1):
+
+ stdout_transport = transport.get_pipe_transport(1)
+ if stdout_transport is not None:
 self.stdout = streams.StreamReader(limit=self._limit,
 loop=self._loop)
- if transport.get_pipe_transport(2):
+ self.stdout.set_transport(stdout_transport)
+
+ stderr_transport = transport.get_pipe_transport(2)
+ if stderr_transport is not None:
 self.stderr = streams.StreamReader(limit=self._limit,
 loop=self._loop)
- stdin = transport.get_pipe_transport(0)
- if stdin is not None:
- self.stdin = streams.StreamWriter(stdin,
+ self.stderr.set_transport(stderr_transport)
+
+ stdin_transport = transport.get_pipe_transport(0)
+ if stdin_transport is not None:
+ self.stdin = streams.StreamWriter(stdin_transport,
 protocol=self,
 reader=None,
 loop=self._loop)
diff --git a/Lib/test/test_asyncio/test_subprocess.py b/Lib/test/test_asyncio/test_subprocess.py
--- a/Lib/test/test_asyncio/test_subprocess.py
+++ b/Lib/test/test_asyncio/test_subprocess.py
@@ -4,6 +4,7 @@
 import signal
 import sys
 import unittest
+from unittest import mock
 from test import support
 if sys.platform != 'win32':
 from asyncio import unix_events
@@ -161,6 +162,37 @@
 self.loop.run_until_complete(proc.communicate(large_data))
 self.loop.run_until_complete(proc.wait())
 
+ def test_pause_reading(self):
+ @asyncio.coroutine
+ def test_pause_reading():
+ limit = 100
+
+ code = '\n'.join((
+ 'import sys',
+ 'sys.stdout.write("x" * %s)' % (limit * 2 + 1),
+ 'sys.stdout.flush()',
+ ))
+ proc = yield from asyncio.create_subprocess_exec(
+ sys.executable, '-c', code,
+ stdin=asyncio.subprocess.PIPE,
+ stdout=asyncio.subprocess.PIPE,
+ limit=limit,
+ loop=self.loop)
+ stdout_transport = proc._transport.get_pipe_transport(1)
+ stdout_transport.pause_reading = mock.Mock()
+
+ yield from proc.wait()
+
+ # The child process produced more than limit bytes of output,
+ # the stream reader transport should pause the protocol to not
+ # allocate too much memory.
+ return stdout_transport.pause_reading.called
+
+ # Issue #22685: Ensure that the stream reader pauses the protocol
+ # when the child process produces too much data
+ called = self.loop.run_until_complete(test_pause_reading())
+ self.assertTrue(called)
+
 
 if sys.platform != 'win32':
 # Unix
-- 
Repository URL: https://hg.python.org/cpython
