python - read - subprocess run cat




Pythonのサブプロセスreadlineのタイムアウト (5)

Python 3では、タイムアウトオプションがサブプロセスモジュールに追加されました。 のような構造を使う

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

analyze(o)

適切な解決策になります。

出力には改行文字が含まれていることが予想されるので、それがテキストであると仮定することは安全です(この場合、 universal_newlines=Trueフラグを強くお勧めします)。

Python2が必須の場合は、 https://pypi.python.org/pypi/subprocess32/ //pypi.python.org/pypi/subprocess32/(バックポート)

純粋なPython Python 2のソリューションについては、 モジュール 'サブプロセス'をタイムアウトとともに使用するを見てください。

私は小さな問題を抱えています。 ここには最小の例があります:

私が持っているもの

scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
while(some_criterium):
    line = scan_process.stdout.readline()
    some_criterium = do_something(line)

私が望むもの

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

私はサブプロセスから行を読み、それを使って何かを行います。 私が望むのは、一定の時間間隔の後に到着した行がなければ終了することです。 どんな勧告?


signal.alarmを使ってみてください:

#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

それが動作することを示すためにいくつかの実行:

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

より詳細な例については、 pGuides参照してください。


あなたの(トムの)ソリューションは動作しselect()が、 Cイディオムでのselect()使用はよりコンパクトです。 これはあなたの答えと同じです

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]

残りは同じです。

pydoc select.select参照してください。

[注意:これは、他の回答のいくつかと同様、Unix特有のものです。]

[注2:編集してOPリクエストごとにラインバッファリングを追加]

[注3:ラインバッファリングはすべての状況において信頼できるものではなく、readline()のブロックにつながる可能性があります]


すべての答えをありがとう! 私は単にselect.pollを使ってstdoutを覗いて問題を解決する方法を見つけました。

import select
...
scan_process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
poll_obj = select.poll()
poll_obj.register(scan_process.stdout, select.POLLIN)   
while(some_criterium and not time_limit):
    poll_result = poll_obj.poll(0)
    if poll_result:
        line = scan_process.stdout.readline()
        some_criterium = do_something(line)
    update(time_limit)

移植可能な解決策は、行の読み取りに時間がかかりすぎると、スレッドを使用して子プロセスを強制終了させることです。

#!/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()

WatchdogTimerクラスはthreading.Timer似ています。再起動またはブロックできる:

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






subprocess