using `p = create_subprocess_exec(..., stdout=subprocess.PIPE, limit=...)`, 
p.stdout has not transport set, so the underlying protocol is unable to pause the reading of the transport, resulting in high memory usage when slowly consuming input from p.stdout, even if the limit parameter is passed. 

A workaround is to set the transport manually after creating the subprocess:
`p.stdout.set_transport(p._transport.get_pipe_transport(1))`, but this should happen inside the create_subprocess call.

> p = create_subprocess_exec(..., stdout=subprocess.PIPE, limit=...)

I don't see "yield from". Do you wait until the coroutine completes?

See the documentation:
https://docs.python.org/dev/library/asyncio-subprocess.html#subprocess-using-streams

Please write a more complete example to demontrate your issue.

Sorry for the confusion, yes i do the yield from. The stdout stream for the process is actually producing data as it should. The subprocess produces a high amount of data (pbzip2), but is only consumed slowly. 

Normally when the buffer limit is reached for a stream reader, it calls pause_reading on the transport inside the feed_data method (see https://code.google.com/p/tulip/source/browse/asyncio/streams.py#365),
but here this is not happening, as the returned reader has no transport set (p.stdout._transport == None). So it fills up all the memory.

Hm... It does look like there's nothing that tells stdout (which is a
StreamReader) about its transport. Wabu, could you experiment with a change
to asyncio/subprocess.py where SubprocessStreamProtocol.connection_made()
calls self.stdout.set_transport(transport) right after creating self.stdout?

On Tue, Oct 21, 2014 at 12:55 PM, wabu <report@bugs.python.org> wrote:

>
> wabu added the comment:
>
> Sorry for the confusion, yes i do the yield from. The stdout stream for
> the process is actually producing data as it should. The subprocess
> produces a high amount of data (pbzip2), but is only consumed slowly.
>
> Normally when the buffer limit is reached for a stream reader, it calls
> pause_reading on the transport inside the feed_data method (see
> https://code.google.com/p/tulip/source/browse/asyncio/streams.py#365),
> but here this is not happening, as the returned reader has no transport
> set (p.stdout._transport == None). So it fills up all the memory.
>
> ----------
>
> _______________________________________
> Python tracker <report@bugs.python.org>
> <http://bugs.python.org/issue22685>
> _______________________________________
>

Here's a more complete example

@coroutine
put_data(filename, queue, chunksize=16000):
    pbzip2 = yield from asyncio.create_subprocess_exec(
                 'pbzip2', '-cd', filename, 
                 stdout=asyncio.subprocess.PIPE, limit=self.chunksize*2)

    while not pbzip2.stdout.at_eof():
        data = yield from pbzip2.stdout.read(chunksize)
        yield from queue.put(data)

adding the workaround after createing the stream fixes the issue:
pbzip2.stdout.set_transport(pbzip2._transport.get_pipe_transport(1))

Can you confirm that this patch fixes the problem (without you needing the workaround in your own code)?

thanks a lot, the fix works!

On 21.10.2014 22:16, Guido van Rossum wrote:
> 
> Guido van Rossum added the comment:
> 
> Can you confirm that this patch fixes the problem (without you needing the workaround in your own code)?
> 
> ----------
> keywords: +patch
> Added file: http://bugs.python.org/file36989/fix22685.patch
> 
> _______________________________________
> Python tracker <report@bugs.python.org>
> <http://bugs.python.org/issue22685>
> _______________________________________
>

Victor, do you think this needs a unittest? It seems kind of difficult to
test for whether memory fills up (the machine may get wedged if it does :-).

On Tue, Oct 21, 2014 at 1:39 PM, wabu <report@bugs.python.org> wrote:

>
> wabu added the comment:
>
> thanks a lot, the fix works!
>
> On 21.10.2014 22:16, Guido van Rossum wrote:
> >
> > Guido van Rossum added the comment:
> >
> > Can you confirm that this patch fixes the problem (without you needing
> the workaround in your own code)?
> >
> > ----------
> > keywords: +patch
> > Added file: http://bugs.python.org/file36989/fix22685.patch
> >
> > _______________________________________
> > Python tracker <report@bugs.python.org>
> > <http://bugs.python.org/issue22685>
> > _______________________________________
> >
>
> ----------
>
> _______________________________________
> Python tracker <report@bugs.python.org>
> <http://bugs.python.org/issue22685>
> _______________________________________
>

Oh ok, now I understood the issue :-)

Attached patch should fix it. It implements Guido's suggestion. There is not unit test yet.

test_pause_reading.py: manual test for pause reading. It should pause reading stdout pipe 4 times on UNIX.

Ok, I just missed completly the Guido posted a patch before me... Maybe we wrote the patch at the same time, I didn't refresh my browser :-)

> Victor, do you think this needs a unittest? It seems kind of difficult to
test for whether memory fills up (the machine may get wedged if it does :-).

Since the pause/resume thing is very tricky, it would be much better to have a unit test. We can probably mock a lot of things, we just need to check that the stream reader calls transport.pause_reading() when feed_data() is called with "too much" data.

On 21.10.2014 22:41, Guido van Rossum wrote:
> Guido van Rossum added the comment:
> 
> Victor, do you think this needs a unittest? It seems kind of difficult to
> test for whether memory fills up (the machine may get wedged if it does :-).

You could setup a the subprocess with
asyncio.async(asyncio.create_subprocess_exec(...)) and then let the
asyncio loop run for a limited time with
loop.run_until_complete(asyncio.sleep(.1)), watching carefully for
higher memory usage after each sleep.

But still it's difficult to create a reliable unit-test with this ...

Victor, your fix is identical to mine except for the variable names. I like your version better. :-)

So now it's just about the unittest.

Here is a patch with a simple unit test.

I chose to modify the pause_reading method of the transport instead of mocking "everything" to test the real code (have a better code coverage).

> Here is a patch with a simple unit test.

Can someone review it? If not, I will commit it without review.

Sorry, looks good.

On Tue, Nov 25, 2014 at 6:37 AM, STINNER Victor <report@bugs.python.org>
wrote:

>
> STINNER Victor added the comment:
>
> > Here is a patch with a simple unit test.
>
> Can someone review it? If not, I will commit it without review.
>
> ----------
>
> _______________________________________
> Python tracker <report@bugs.python.org>
> <http://bugs.python.org/issue22685>
> _______________________________________
>

New changeset f75d40c02f0a by Victor Stinner in branch '3.4':
Closes #22685, asyncio: Set the transport of stdout and stderr StreamReader
https://hg.python.org/cpython/rev/f75d40c02f0a

New changeset 7da2288183d1 by Victor Stinner in branch 'default':
(Merge 3.4) Closes #22685, asyncio: Set the transport of stdout and stderr
https://hg.python.org/cpython/rev/7da2288183d1

Fix also pushed to Python 3.4, 3.5 and to Tulip. Thanks for the report wabu.

Tulip commit:

changeset:   1350:c3a9d355eb34
user:        Victor Stinner <victor.stinner@gmail.com>
date:        Tue Nov 25 17:17:13 2014 +0100
files:       asyncio/subprocess.py tests/test_subprocess.py
description:
Python issue #22685: 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.

thanks for the fixes 'n' integration

You can workaround the issue by setting manually the transport on the StreamReader objects. Tell me if you need help to write the workaround if you cannot wait for a release containing the fix.

Buildbot failures observed on koobs-freebsd9 and koobs-freebsd10 for 3.x and 3.4, respectively.

Both logs are attached, inlined failure below:

======================================================================
FAIL: test_pause_reading (test.test_asyncio.test_subprocess.SubprocessSafeWatcherTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/home/buildbot/python/3.4.koobs-freebsd10/build/Lib/test/test_asyncio/test_subprocess.py", line 194, in test_pause_reading
    self.assertTrue(called)
AssertionError: False is not true

======================================================================

> Buildbot failures observed on koobs-freebsd9 and koobs-freebsd10 for 3.x and 3.4, respectively.

I'm unable to reproduce the failure with "./python -m test -F test_asyncio" or "./python -m test -F -m test_pause_reading test_asyncio" on my FreeBSD 9.1 VM.

New changeset 737355f61ba2 by Victor Stinner in branch '3.4':
Issue #22685: Debug test_pause_reading() on FreeBSD
https://hg.python.org/cpython/rev/737355f61ba2

New changeset 0dd91298eb17 by Victor Stinner in branch 'default':
Issue #22685, asyncio: mock also resume_reading in test_pause_reading()
https://hg.python.org/cpython/rev/0dd91298eb17

New changeset 276515d2ceed by Victor Stinner in branch 'default':
Issue #22685, asyncio: resume_reading() must also be called in test_pause_reading()
https://hg.python.org/cpython/rev/276515d2ceed

> Buildbot failures observed on koobs-freebsd9 and koobs-freebsd10 for 3.x and 3.4, respectively.

It looks like test_asyncio pass on the last 5 builds of the following buildbots, and so I consider the issue as closed.

http://buildbot.python.org/all/builders/AMD64 FreeBSD 9.x 3.x
http://buildbot.python.org/all/builders/AMD64%20FreeBSD%2010.0%203.x

New changeset 2c18dd99829c by Victor Stinner in branch '3.4':
Issue #22685: Fix test_pause_reading() of asyncio test_subprocess
https://hg.python.org/cpython/rev/2c18dd99829c

New changeset 992ce0dcfb29 by Victor Stinner in branch '3.4':
Issue #22685: Fix test_pause_reading() of asyncio/test_subprocess
https://hg.python.org/cpython/rev/992ce0dcfb29