Une lecture non bloquante sur un subprocess.PIPE en Python

Question

Je utilise le module subprocess pour démarrer un sous-processus et me connecter à son flux de sortie (sortie standard). Je veux être capable d'effectuer des lectures non bloquantes sur sa sortie standard. Y a-t-il un moyen de rendre .readline non bloquant ou de vérifier s'il y a des données sur le flux avant d'invoquer .readline? Je voudrais que cela soit portable ou au moins fonctionne sous Windows et Linux.

Voici comment je le fais actuellement (cela bloque sur le .readline si aucune donnée n'est disponible) :

p = subprocess.Popen('monprogramme.exe', stdout = subprocess.PIPE)
output_str = p.stdout.readline()

Answer 1

fcntl, select, asyncproc ne seront pas utiles dans ce cas.

Une façon fiable de lire un flux sans bloquer indépendamment du système d'exploitation est d'utiliser Queue.get_nowait():

import sys
from subprocess import PIPE, Popen
from threading  import Thread

try:
    from queue import Queue, Empty
except ImportError:
    from Queue import Queue, Empty  # python 2.x

ON_POSIX = 'posix' in sys.builtin_module_names

def enqueue_output(out, queue):
    for line in iter(out.readline, b''):
        queue.put(line)
    out.close()

p = Popen(['myprogram.exe'], stdout=PIPE, bufsize=1, close_fds=ON_POSIX)
q = Queue()
t = Thread(target=enqueue_output, args=(p.stdout, q))
t.daemon = True # thread dies with the program
t.start()

# ... do other things here

# read line without blocking
try:  line = q.get_nowait() # or q.get(timeout=.1)
except Empty:
    print('no output yet')
else: # got line
    # ... do something with line

Answer 2

J'ai souvent eu un problème similaire; les programmes Python que j'écris ont fréquemment besoin de pouvoir exécuter une fonctionnalité principale tout en acceptant simultanément l'entrée de l'utilisateur depuis la ligne de commande (stdin). Mettre simplement la fonctionnalité de gestion de l'entrée utilisateur dans un autre thread ne résout pas le problème car readline() bloque et n'a pas de délai d'attente. Si la fonctionnalité principale est terminée et qu'il n'y a plus besoin d'attendre une autre entrée de l'utilisateur, je veux généralement que mon programme se termine, mais il ne le peut pas car readline() bloque toujours dans l'autre thread en attendant une ligne. Une solution que j'ai trouvée à ce problème est de rendre stdin un fichier non bloquant en utilisant le module fcntl :

import fcntl
import os
import sys

# rendre stdin un fichier non bloquant
fd = sys.stdin.fileno()
fl = fcntl.fcntl(fd, fcntl.F_GETFL)
fcntl.fcntl(fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)

# thread de gestion de l'entrée utilisateur
while mainThreadIsRunning:
      try: input = sys.stdin.readline()
      except: continue
      handleInput(input)

À mon avis, c'est un peu plus propre que d'utiliser les modules select ou signal pour résoudre ce problème mais cela ne fonctionne que sur UNIX...

Answer 3

Python 3.4 introduit une nouvelle API provisoire pour l'IO asynchrone -- module asyncio.

L'approche est similaire à la réponse basée sur twisted de @Bryan Ward -- définir un protocole et ses méthodes sont appelées dès que les données sont prêtes :

#!/usr/bin/env python3
import asyncio
import os

class SubprocessProtocol(asyncio.SubprocessProtocol):
    def pipe_data_received(self, fd, data):
        if fd == 1: # données stdout reçues (bytes)
            print(data)

    def connection_lost(self, exc):
        loop.stop() # arrêter loop.run_forever()

if os.name == 'nt':
    loop = asyncio.ProactorEventLoop() # pour les pipes des sous-processus sur Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(loop.subprocess_exec(SubprocessProtocol, 
        "myprogram.exe", "arg1", "arg2"))
    loop.run_forever()
finally:
    loop.close()

Voir "Sous-processus" dans la documentation.

Il existe une interface de haut niveau asyncio.create_subprocess_exec() qui renvoie des objets Process qui permettent de lire une ligne de manière asynchrone en utilisant la coroutine StreamReader.readline() (avec la syntaxe async/await Python 3.5+):

#!/usr/bin/env python3.5
import asyncio
import locale
import sys
from asyncio.subprocess import PIPE
from contextlib import closing

async def readline_and_kill(*args):
    # démarrer le processus enfant
    process = await asyncio.create_subprocess_exec(*args, stdout=PIPE)

    # lire la ligne (séquence d'octets se terminant par b'\n') de manière asynchrone
    async for line in process.stdout:
        print("ligne reçue :", line.decode(locale.getpreferredencoding(False)))
        break
    process.kill()
    return await process.wait() # attendre que le processus enfant se termine

if sys.platform == "win32":
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

with closing(loop):
    sys.exit(loop.run_until_complete(readline_and_kill(
        "myprogram.exe", "arg1", "arg2")))

readline_and_kill() réalise les tâches suivantes :

• démarrer le sous-processus, rediriger son stdout vers un pipe
• lire une ligne du stdout du sous-processus de manière asynchrone
• tuer le sous-processus
• attendre sa sortie

Chaque étape pourrait être limitée par des secondes de timeout si nécessaire.

Answer 4

Essayez le module asyncproc. Par exemple :

import os
from asyncproc import Process
myProc = Process("myprogram.app")

while True:
    # vérifier si le processus est terminé
    poll = myProc.wait(os.WNOHANG)
    if poll != None:
        break
    # imprimer toute nouvelle sortie
    out = myProc.read()
    if out != "":
        print out

Le module s'occupe de tout le threading, comme suggéré par S.Lott.

Answer 5

Vous pouvez faire cela très facilement dans Twisted. En fonction de votre code existant, cela peut ne pas être si simple à utiliser, mais si vous construisez une application Twisted, des choses comme celles-ci deviennent presque triviales. Vous créez une classe ProcessProtocol, et remplacez la méthode outReceived(). Twisted (en fonction du réacteur utilisé) n'est généralement qu'une grande boucle select() avec des rappels installés pour gérer les données provenant de différents descripteurs de fichiers (souvent des sockets réseau). Ainsi, la méthode outReceived() consiste simplement à installer un rappel pour gérer les données provenant de STDOUT. Un exemple simple démontrant ce comportement est le suivant :

from twisted.internet import protocol, reactor

class MyProcessProtocol(protocol.ProcessProtocol):

    def outReceived(self, data):
        print data

proc = MyProcessProtocol()
reactor.spawnProcess(proc, './myprogram', ['./myprogram', 'arg1', 'arg2', 'arg3'])
reactor.run()

La documentation Twisted contient des informations intéressantes à ce sujet.

Si vous construisez toute votre application autour de Twisted, cela rend la communication asynchrone avec d'autres processus, locaux ou distants, vraiment élégante comme ceci. En revanche, si votre programme n'est pas construit sur Twisted, cela ne sera pas vraiment utile. Espérons que cela puisse être utile à d'autres lecteurs, même si ce n'est pas applicable à votre application particulière.