Note, that the data read is buffered in memory, so do not use
this method if the data size is large or unlimited.
+ If this coroutine is cancelled (for example, when a timeout is
+ set with :func:`~asyncio.wait_for`), the output that was already
+ read is not lost: call :meth:`!communicate` again to read the
+ remaining output and get the complete data::
+
+ try:
+ stdout, stderr = await asyncio.wait_for(
+ proc.communicate(), timeout=5.0)
+ except TimeoutError:
+ proc.kill()
+ stdout, stderr = await proc.communicate()
+
+ Passing *input* after a previous :meth:`!communicate` call was
+ cancelled raises :exc:`ValueError`; pass ``input=None`` to
+ continue the communication, the original *input* is used.
+
.. versionchanged:: 3.12
*stdin* gets closed when ``input=None`` too.
+ .. versionchanged:: next
+
+ If :meth:`!communicate` is cancelled, the output that was
+ already read is now preserved and returned by a subsequent
+ :meth:`!communicate` call. Passing *input* to a
+ :meth:`!communicate` call following a cancelled one now raises
+ :exc:`ValueError`.
+
.. method:: send_signal(signal)
Sends the signal *signal* to the child process.
self.stdout = protocol.stdout
self.stderr = protocol.stderr
self.pid = transport.get_pid()
+ self._communication_started = False
+ self._input = None
+ self._input_written = False
+ self._stdout_buf = bytearray()
+ self._stderr_buf = bytearray()
def __repr__(self):
return f'<{self.__class__.__name__} {self.pid}>'
async def _feed_stdin(self, input):
debug = self._loop.get_debug()
try:
- if input is not None:
+ if input is not None and not self._input_written:
self.stdin.write(input)
+ self._input_written = True
if debug:
logger.debug(
'%r communicate: feed stdin (%s bytes)', self, len(input))
transport = self._transport.get_pipe_transport(fd)
if fd == 2:
stream = self.stderr
+ buf = self._stderr_buf
else:
assert fd == 1
stream = self.stdout
+ buf = self._stdout_buf
if self._loop.get_debug():
name = 'stdout' if fd == 1 else 'stderr'
logger.debug('%r communicate: read %s', self, name)
- output = await stream.read()
+ # Append each block to the persistent buffer as soon as it is
+ # read so that no output is lost if this coroutine is cancelled.
+ while block := await stream.read(stream._limit):
+ buf += block
if self._loop.get_debug():
name = 'stdout' if fd == 1 else 'stderr'
logger.debug('%r communicate: close %s', self, name)
transport.close()
- return output
async def communicate(self, input=None):
+ if self._communication_started:
+ if input:
+ raise ValueError(
+ "Cannot send input after starting communication")
+ else:
+ self._input = input
+ self._communication_started = True
if self.stdin is not None:
- stdin = self._feed_stdin(input)
+ stdin = self._feed_stdin(self._input)
else:
stdin = self._noop()
if self.stdout is not None:
stderr = self._read_stream(2)
else:
stderr = self._noop()
- stdin, stdout, stderr = await tasks.gather(stdin, stdout, stderr)
+ await tasks.gather(stdin, stdout, stderr)
await self.wait()
+ if self.stdout is not None:
+ stdout = self._stdout_buf.take_bytes()
+ else:
+ stdout = None
+ if self.stderr is not None:
+ stderr = self._stderr_buf.take_bytes()
+ else:
+ stderr = None
return (stdout, stderr)
self.assertEqual(exitcode, 0)
self.assertEqual(stdout, b'')
+ def test_communicate_cancelled_mid_read_retry(self):
+ # gh-139373: output read before communicate() was cancelled must
+ # be returned by a subsequent communicate() call.
+ code = ('import sys, time;'
+ 'sys.stdout.buffer.write(b"first\\n");'
+ 'sys.stdout.buffer.flush();'
+ 'time.sleep(3600)')
+
+ async def run():
+ proc = await asyncio.create_subprocess_exec(
+ sys.executable, '-c', code,
+ stdout=subprocess.PIPE,
+ )
+ task = asyncio.create_task(proc.communicate())
+ # wait until communicate() has read the first line
+ while not proc._stdout_buf:
+ await asyncio.sleep(0.01)
+ task.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await task
+ proc.kill()
+ stdout, stderr = await proc.communicate()
+ return stdout
+
+ task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+ stdout = self.loop.run_until_complete(task)
+ self.assertEqual(stdout, b'first\n')
+
+ def test_communicate_cancelled_during_wait_retry(self):
+ # gh-139373: cancellation landing after the output was fully read
+ # but before wait() completed must not lose the output.
+ code = ('import os, time;'
+ 'os.write(1, b"all output\\n");'
+ 'os.close(1);'
+ 'time.sleep(3600)')
+
+ async def run():
+ proc = await asyncio.create_subprocess_exec(
+ sys.executable, '-c', code,
+ stdout=subprocess.PIPE,
+ )
+ task = asyncio.create_task(proc.communicate())
+ # wait until the stdout reader has consumed everything up to
+ # EOF; communicate() is then blocked on wait()
+ while not proc.stdout.at_eof():
+ await asyncio.sleep(0.01)
+ task.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await task
+ proc.kill()
+ stdout, stderr = await proc.communicate()
+ return stdout
+
+ task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+ stdout = self.loop.run_until_complete(task)
+ self.assertEqual(stdout, b'all output\n')
+
+ def test_communicate_cancelled_input_not_resendable(self):
+ # gh-139373: like subprocess.Popen.communicate(), sending new
+ # input after a cancelled communicate() call is an error.
+ async def run():
+ proc = await asyncio.create_subprocess_exec(
+ *PROGRAM_CAT,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ )
+ task = asyncio.create_task(proc.communicate(b'data'))
+ await asyncio.sleep(0)
+ task.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await task
+ with self.assertRaises(ValueError):
+ await proc.communicate(b'more data')
+ proc.kill()
+ await proc.communicate()
+
+ self.loop.run_until_complete(
+ asyncio.wait_for(run(), support.LONG_TIMEOUT))
+
+ def test_communicate_cancelled_stdin_retry(self):
+ # gh-139373: input already fed before cancellation is not re-sent
+ # by the retried communicate() call, and the output is preserved.
+ code = ('import sys, time;'
+ 'sys.stdout.buffer.write(sys.stdin.buffer.read());'
+ 'sys.stdout.buffer.flush();'
+ 'time.sleep(3600)')
+
+ async def run():
+ proc = await asyncio.create_subprocess_exec(
+ sys.executable, '-c', code,
+ stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE,
+ )
+ task = asyncio.create_task(proc.communicate(b'hello'))
+ # the child echoes stdin only after it is closed, so once
+ # output arrives the input was fully written and
+ # communicate() is blocked on wait()
+ while not proc._stdout_buf:
+ await asyncio.sleep(0.01)
+ task.cancel()
+ with self.assertRaises(asyncio.CancelledError):
+ await task
+ proc.kill()
+ stdout, stderr = await proc.communicate()
+ return stdout
+
+ task = asyncio.wait_for(run(), support.LONG_TIMEOUT)
+ stdout = self.loop.run_until_complete(task)
+ self.assertEqual(stdout, b'hello')
+
def test_shell(self):
proc = self.loop.run_until_complete(
asyncio.create_subprocess_shell('exit 7')
projects / thirdparty / Python / cpython.git / commitdiff
