Connect zwei Prozesse gestartet mit asyncio.subprocess.create_subprocess_exec()

Question

Wenn mit der alten Schule subprocess.Popen() API zwei Prozesse gestartet wird, kann ich leicht anschließen Standard aus von einem Prozess zur Standard in eines anderen Prozess, eine Pipeline zu schaffen, in die gleiche Weise wie ein UNIX-Shell tun wird, wenn Befehle mit | Verbindungs:

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() 

wie kann ich das gleiche zu erreichen, wenn den asynchronen API von asyncio.subprocess.create_subprocess_exec() (oder ähnlichem) unter Verwendung von? Das ist, was ich versucht:

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()) 

Aber der zweite Aufruf von create_subprocess_exec() beschwert sich, dass die zu stdin gebene Argument hat keine fileno (was stimmt):

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' 

Wie kann ich das gleiche Ergebnis wie im synchronen Beispiel oben?

Answer 1

In asyncio ist process.stdout eigentlich ein StreamReader, kein Dateiobjekt. Auf das Dateiobjekt kann über process._transport._proc.stdout zugegriffen werden. Leider können Sie es nicht verwenden, da es bereits in der Ereignisschleife registriert wurde, um die Stream-Schnittstelle process.stdout bereitzustellen.

Ein Weg, mit dem Problem umzugehen ist, Ihr eigenes Rohr zu erstellen und die Datei-Deskriptoren zum subprocess passieren:

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() 

Beachten Sie, dass die write Dateideskriptors explizit geschlossen werden sollen, sobald der erste Teilprozess gestartet wird (Es wird nicht automatisch geschlossen, es sei denn, Sie verwenden subprocess.PIPE). Der Dateideskriptor read muss ebenfalls geschlossen werden, wie erklärt here.