timeout sur la ligne de lecture des sous-processus en python

Voici une solution portable qui renforce le délai de lecture d'une ligne simple en utilisant asyncio :

#!/usr/bin/env python3
import asyncio
import sys
from asyncio.subprocess import PIPE, STDOUT

async def run_command(*args, timeout=None):
    # start child process
    # NOTE: universal_newlines parameter is not supported
    process = await asyncio.create_subprocess_exec(*args,
            stdout=PIPE, stderr=STDOUT)

    # read line (sequence of bytes ending with b'\n') asynchronously
    while True:
        try:
            line = await asyncio.wait_for(process.stdout.readline(), timeout)
        except asyncio.TimeoutError:
            pass
        else:
            if not line: # EOF
                break
            elif do_something(line): 
                continue # while some criterium is satisfied
        process.kill() # timeout or some criterium is not satisfied
        break
    return await process.wait() # wait for the child process to exit


if sys.platform == "win32":
    loop = asyncio.ProactorEventLoop() # for subprocess' pipes on Windows
    asyncio.set_event_loop(loop)
else:
    loop = asyncio.get_event_loop()

returncode = loop.run_until_complete(run_command("cmd", "arg 1", "arg 2",
                                                 timeout=10))
loop.close()

j'ai utilisé quelque chose d'un peu plus général en python (L'IIRC a aussi compilé à partir de ces questions, mais je ne me souviens pas lesquelles).

import thread
from threading import Timer

def run_with_timeout(timeout, default, f, *args, **kwargs):
    if not timeout:
        return f(*args, **kwargs)
    try:
        timeout_timer = Timer(timeout, thread.interrupt_main)
        timeout_timer.start()
        result = f(*args, **kwargs)
        return result
    except KeyboardInterrupt:
        return default
    finally:
        timeout_timer.cancel()

soyez averti, cependant, ceci utilise une interruption pour arrêter n'importe quelle fonction que vous lui donnez. Ce n'est peut-être pas une bonne idée pour toutes les fonctions et cela vous empêche également de fermer le programme avec ctrl+c pendant le timeout (c.-à-d. ctrl+c sera manipulé comme un timeout) Vous pouvez utiliser ce un appel comme:

scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
while(some_criterium):
    line = run_with_timeout(timeout, None, scan_process.stdout.readline)
    if line is None:
        break
    else:
        some_criterium = do_something(line)

est peut-être un peu exagéré. Je pense qu'il y a une option plus simple pour votre affaire que je ne connais pas.

en Python 3, une option timeout a été ajoutée au module subprocess. En utilisant une structure comme

try:
    o, e = process.communicate(timeout=10)
except TimeoutExpired:
    process.kill()
    o, e = process.communicate()

analyze(o)

serait une bonne solution.

puisque la sortie est censée contenir un nouveau caractère de ligne, il est sûr de supposer qu'il s'agit d'un texte (comme dans imprimable, lisible), auquel cas le drapeau universal_newlines=True est fortement recommandé.

si Python2 est un must, s'il vous plaît utiliser https://pypi.python.org/pypi/subprocess32 / (backport)

pour une solution Python Python 2 pure, regardez en utilisant le module 'subprocess' avec timeout .

essayez le signal.alarme:

#timeout.py
import signal,sys

def timeout(sig,frm):
  print "This is taking too long..."
  sys.exit(1)

signal.signal(signal.SIGALRM, timeout)
signal.alarm(10)
byte=0

while 'IT' not in open('/dev/urandom').read(2):
  byte+=2
print "I got IT in %s byte(s)!" % byte

quelques passages pour montrer que ça marche:

$ python timeout.py 
This is taking too long...
$ python timeout.py 
I got IT in 4672 byte(s)!

pour un exemple plus détaillé, voir pGuides .

une solution portable est d'utiliser un fil pour tuer le processus enfant si la lecture d'une ligne prend trop de temps:

#!/usr/bin/env python3
from subprocess import Popen, PIPE, STDOUT

timeout = 10
with Popen(command, stdout=PIPE, stderr=STDOUT,
           universal_newlines=True) as process:  # text mode
    # kill process in timeout seconds unless the timer is restarted
    watchdog = WatchdogTimer(timeout, callback=process.kill, daemon=True)
    watchdog.start()
    for line in process.stdout:
        # don't invoke the watcthdog callback if do_something() takes too long
        with watchdog.blocked:
            if not do_something(line):  # some criterium is not satisfied
                process.kill()
                break
            watchdog.restart()  # restart timer just before reading the next line
    watchdog.cancel()

où WatchdogTimer est comme threading.Timer qui peut être redémarré et / ou bloqué:

from threading import Event, Lock, Thread
from subprocess import Popen, PIPE, STDOUT
from time import monotonic  # use time.time or monotonic.monotonic on Python 2

class WatchdogTimer(Thread):
    """Run *callback* in *timeout* seconds unless the timer is restarted."""

    def __init__(self, timeout, callback, *args, timer=monotonic, **kwargs):
        super().__init__(**kwargs)
        self.timeout = timeout
        self.callback = callback
        self.args = args
        self.timer = timer
        self.cancelled = Event()
        self.blocked = Lock()

    def run(self):
        self.restart() # don't start timer until `.start()` is called
        # wait until timeout happens or the timer is canceled
        while not self.cancelled.wait(self.deadline - self.timer()):
            # don't test the timeout while something else holds the lock
            # allow the timer to be restarted while blocked
            with self.blocked:
                if self.deadline <= self.timer() and not self.cancelled.is_set():
                    return self.callback(*self.args)  # on timeout

    def restart(self):
        """Restart the watchdog timer."""
        self.deadline = self.timer() + self.timeout

    def cancel(self):
        self.cancelled.set()

pendant que votre solution (celle de Tom) fonctionne, utiliser select() dans l'idiome C est plus compact. c'est l'équivalent de votre réponse

from select import select
scan_process = subprocess.Popen(command, 
                                stdout=subprocess.PIPE,
                                stderr=subprocess.STDOUT,
                                bufsize=1)  # line buffered
while some_criterium and not time_limit:
    poll_result = select([scan_process.stdout], [], [], time_limit)[0]

le reste est le même.

voir pydoc select.select .

[Note: Ceci est spécifique à Unix, comme le sont certaines des autres réponses.]

[Note 2: modifié pour ajouter un tampon de ligne conformément à la demande OP]

[Note 3: le tampon de ligne peut ne pas être fiable en toutes circonstances, conduisant au blocage readline ()]