Connect two processes started with asyncio.subprocess.create_subprocess_exec()

Question

When starting two processes with the old school subprocess.Popen() API, I can easily connect standard out of one process to standard in of another process, creating a pipeline in the same way as a UNIX shell will do when connecting commands with |:

from subprocess import Popen, PIPE

process_1 = Popen(['ls'], stdout = PIPE)
process_2 = Popen(['wc'], stdin = process_1.stdout)

process_1.wait()
process_2.wait()

How can I accomplish the same when using the asynchronous API from asyncio.subprocess.create_subprocess_exec() (or similar)? This is what I tried:

from asyncio.events import get_event_loop
from asyncio.subprocess import PIPE, create_subprocess_exec

async def main():
    process_1 = await create_subprocess_exec('ls', stdout = PIPE)
    process_2 = await create_subprocess_exec('wc', stdin = process_1.stdout)

    await process_1.wait()
    await process_2.wait()

get_event_loop().run_until_complete(main())

But the second call to create_subprocess_exec() complains that the argument passed to stdin has no fileno (which is true):

Traceback (most recent call last):
  File ".../test-async.py", line 11, in <module>
     get_event_loop().run_until_complete(main())
[...]
  File ".../test-async.py", line 6, in main
     process_2 = await create_subprocess_exec('wc', stdin = process_1.stdout)
[...]
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/subprocess.py", line 1388, in _get_handles
     p2cread = stdin.fileno()
AttributeError: 'StreamReader' object has no attribute 'fileno'

How can I get the same result as in the synchronous example above?

unrelated: you should add process_1.stdout.close() after Popen(['wc'],..). See How do I use subprocess.Popen to connect multiple processes by pipes? — jfs
– jfs, Commented Apr 24, 2016 at 18:06

Vincent · Accepted Answer · 2016-04-24 20:36:25Z

17

In asyncio, process.stdout is actually a StreamReader, not a file object. The file object can be accessed through process._transport._proc.stdout. Unfortunately, you won't be able to use it since it has already been registered in the event loop in order to provide the stream interface process.stdout.

One way to deal with the issue is to create your own pipe and pass the file descriptors to the subprocess:

async def main():
    read, write = os.pipe()
    process_1 = await create_subprocess_exec('ls', stdout=write)
    os.close(write)
    process_2 = await create_subprocess_exec('wc', stdin=read, stdout=PIPE)
    os.close(read)
    return await process_2.stdout.read()

Note that the write file descriptor should be explicitly closed once the first subprocess is started (it is not automatically closed unless you use subprocess.PIPE). The read file descriptor also needs to be closed, as explained here.

edited Apr 24, 2016 at 20:36

answered Apr 16, 2016 at 16:00

Vincent

13.5k1 gold badge51 silver badges68 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

jfs Over a year ago

could you provide a reference for the statement: "it is closed automatically only if you use subprocess.PIPE"? As I understand the reason you want os.close(write) in the parent is so that ls would get SIGPIPE and EPIPE if wc dies prematurely. It is the same for subprocess.PIPE (at least for ordinary Popen() as demonstrated by the code example in the docs)

Vincent Over a year ago

@J.F.Sebastian I noticed it by looking at the code, I don't know if it's documented somewhere. However, your example is about closing the reader, not the writer. In my example, the writer needs to be closed in order to set up the pipe properly (see this answer).

jfs Over a year ago

You are right about the write end of the pipe (it is closed because it is unused in the parent—it would also explain why it is the responsibility of the code that creates the subprocess.PIPE pipe to close it—it is unrelated to asyncio and the corresponding os.close() call is in the subprocess module). The example that I cited would correspond to process_1.stdout.close() above i.e, os.close(read) should be called.

Vincent Over a year ago

@J.F.Sebastian You're right, I didn't realize the reader also needs to be closed. Thanks for your insight, see my edit.

Collectives™ on Stack Overflow

Connect two processes started with asyncio.subprocess.create_subprocess_exec()

1 Answer 1

4 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related