データを返す任意のコマンドを実行するstdoutか、ゼロ以外の終了コードで例外を発生させるPythonコードは次のとおりです。
proc = subprocess.Popen(
cmd,
stderr=subprocess.STDOUT, # Merge stdout and stderr
stdout=subprocess.PIPE,
shell=True)
communicateプロセスが終了するのを待つために使用されます：
stdoutdata, stderrdata = proc.communicate()
モジュールはタイムアウト（subprocessX秒を超えて実行されているプロセスを強制終了する機能）をサポートしていないため、communicate実行に永遠にかかる場合があります。
WindowsとLinuxで実行することを目的としたPythonプログラムでタイムアウトを実装する最も簡単な方法は何ですか？
低レベルの詳細についてはよくわかりません。しかし、Python 2.6では、APIがスレッドを待機してプロセスを終了する機能を提供していることを考えると、別のスレッドでプロセスを実行するのはどうでしょうか。
import subprocess, threading
class Command(object):
def __init__(self, cmd):
self.cmd = cmd
self.process = None
def run(self, timeout):
def target():
print 'Thread started'
self.process = subprocess.Popen(self.cmd, shell=True)
self.process.communicate()
print 'Thread finished'
thread = threading.Thread(target=target)
thread.start()
thread.join(timeout)
if thread.is_alive():
print 'Terminating process'
self.process.terminate()
thread.join()
print self.process.returncode
command = Command("echo 'Process started'; sleep 2; echo 'Process finished'")
command.run(timeout=3)
command.run(timeout=1)
私のマシンでのこのスニペットの出力は次のとおりです。
Thread started
Process started
Process finished
Thread finished
0
Thread started
Process started
Terminating process
Thread finished
-15
ここで、最初の実行ではプロセスが正しく終了し（戻りコード0）、2番目の実行ではプロセスが終了した（戻りコード-15）ことがわかります。
私はWindowsでテストしていません。ただし、サンプルコマンドを更新する以外は、thread.joinまたはprocess.terminateがサポートされていないことを示すドキュメントが見つからないため、機能するはずです。
Python 3.3以降：
from subprocess import STDOUT, check_output
output = check_output(cmd, stderr=STDOUT, timeout=seconds)
outputコマンドのマージされたstdout、stderrデータを含むバイト文字列です。
check_outputメソッドCalledProcessErrorとは異なり、質問のテキストで指定されているように、ゼロ以外の終了ステータスで発生しますproc.communicate()。
shell=True不必要に使われることが多いので削除しました。cmd本当に必要な場合は、いつでも追加できます。shell=Trueつまり、子プロセスが独自の子孫を生成する場合。check_output()タイムアウトが示すよりもはるかに遅く戻る可能性があります。サブプロセスのタイムアウトの失敗を参照してください。
タイムアウト機能はsubprocess32、Python2.xで3.2以降のサブプロセスモジュールのバックポートを介して利用できます。
jcolladoの答えは、 threading.Timerクラスを使用して簡略化できます。
import shlex
from subprocess import Popen, PIPE
from threading import Timer
def run(cmd, timeout_sec):
proc = Popen(shlex.split(cmd), stdout=PIPE, stderr=PIPE)
timer = Timer(timeout_sec, proc.kill)
try:
timer.start()
stdout, stderr = proc.communicate()
finally:
timer.cancel()
# Examples: both take 1 second
run("sleep 1", 5) # process ends normally at 1 second
run("sleep 5", 1) # timeout happens at 1 second
Unixを使用している場合は、
import signal
...
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
signal.signal(signal.SIGALRM, alarm_handler)
signal.alarm(5*60) # 5 minutes
try:
stdoutdata, stderrdata = proc.communicate()
signal.alarm(0) # reset the alarm
except Alarm:
print "Oops, taking too long!"
# whatever else
これは、適切なプロセスキルを備えたモジュールとしてのAlexMartelliのソリューションです。他のアプローチは、proc.communicate（）を使用しないため、機能しません。したがって、大量の出力を生成するプロセスがある場合、そのプロセスはその出力バッファーをいっぱいにしてから、そこから何かを読み取るまでブロックします。
from os import kill
from signal import alarm, signal, SIGALRM, SIGKILL
from subprocess import PIPE, Popen
def run(args, cwd = None, shell = False, kill_tree = True, timeout = -1, env = None):
'''
Run a command with a timeout after which it will be forcibly
killed.
'''
class Alarm(Exception):
pass
def alarm_handler(signum, frame):
raise Alarm
p = Popen(args, shell = shell, cwd = cwd, stdout = PIPE, stderr = PIPE, env = env)
if timeout != -1:
signal(SIGALRM, alarm_handler)
alarm(timeout)
try:
stdout, stderr = p.communicate()
if timeout != -1:
alarm(0)
except Alarm:
pids = [p.pid]
if kill_tree:
pids.extend(get_process_children(p.pid))
for pid in pids:
# process might have died before getting to this line
# so wrap to avoid OSError: no such process
try:
kill(pid, SIGKILL)
except OSError:
pass
return -9, '', ''
return p.returncode, stdout, stderr
def get_process_children(pid):
p = Popen('ps --no-headers -o pid --ppid %d' % pid, shell = True,
stdout = PIPE, stderr = PIPE)
stdout, stderr = p.communicate()
return [int(p) for p in stdout.split()]
if __name__ == '__main__':
print run('find /', shell = True, timeout = 3)
print run('find', shell = True)
timeoutサブプロセスモジュールでサポートされるようにcall()なりcommunicate()ました（Python3.3以降）。
import subprocess
subprocess.call("command", timeout=20, shell=True)
これにより、コマンドが呼び出され、例外が発生します
subprocess.TimeoutExpired
コマンドが20秒後に終了しない場合。
次に、例外を処理して、次のようにコードを続行できます。
try:
subprocess.call("command", timeout=20, shell=True)
except subprocess.TimeoutExpired:
# insert code here
お役に立てれば。
誰も使用について言及していないことに驚いたtimeout
timeout 5 ping -c 3 somehost
これは明らかにすべてのユースケースで機能するわけではありませんが、単純なスクリプトを扱っている場合、これに勝るものはありません。
Macユーザー向けのcoreutilsでgtimeoutとしても利用できhomebrewます。
Python 3.5以降、新しいsubprocess.runユニバーサルコマンド（check_call、check_output...を置き換えることを目的としています）があり、timeout=パラメーターもあります。
サブプロセス。run（args、*、stdin = None、input = None、stdout = None、stderr = None、shell = False、cwd = None、timeout = None、check = False、encoding = None、errors = None）
argsで記述されたコマンドを実行します。コマンドが完了するのを待ってから、
CompletedProcessインスタンスを返します。
subprocess.TimeoutExpiredタイムアウトが経過すると例外が発生します。
ススーディオの答えを変更しました。関数は次を返します：（returncode、、、）-そしてutf-8文字列にデコードされますstdoutstderrtimeoutstdoutstderr
def kill_proc(proc, timeout):
timeout["value"] = True
proc.kill()
def run(cmd, timeout_sec):
proc = subprocess.Popen(shlex.split(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
timeout = {"value": False}
timer = Timer(timeout_sec, kill_proc, [proc, timeout])
timer.start()
stdout, stderr = proc.communicate()
timer.cancel()
return proc.returncode, stdout.decode("utf-8"), stderr.decode("utf-8"), timeout["value"]
もう1つのオプションは、communicate（）でポーリングする代わりに、一時ファイルに書き込んでstdoutのブロックを防ぐことです。これは、他の答えがうまくいかなかった私にとってはうまくいきました。たとえば、Windowsで。
outFile = tempfile.SpooledTemporaryFile()
errFile = tempfile.SpooledTemporaryFile()
proc = subprocess.Popen(args, stderr=errFile, stdout=outFile, universal_newlines=False)
wait_remaining_sec = timeout
while proc.poll() is None and wait_remaining_sec > 0:
time.sleep(1)
wait_remaining_sec -= 1
if wait_remaining_sec <= 0:
killProc(proc.pid)
raise ProcessIncompleteError(proc, timeout)
# read temp streams from start
outFile.seek(0);
errFile.seek(0);
out = outFile.read()
err = errFile.read()
outFile.close()
errFile.close()
jcolladoPythonモジュールeasyprocessにスレッド化したソリューションを追加しました。
インストール：
pip install easyprocess
例：
from easyprocess import Proc
# shell is not supported!
stdout=Proc('ping localhost').call(timeout=1.5).stdout
print stdout
これが私の解決策です。私はスレッドとイベントを使用していました。
import subprocess
from threading import Thread, Event
def kill_on_timeout(done, timeout, proc):
if not done.wait(timeout):
proc.kill()
def exec_command(command, timeout):
done = Event()
proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
watcher = Thread(target=kill_on_timeout, args=(done, timeout, proc))
watcher.daemon = True
watcher.start()
data, stderr = proc.communicate()
done.set()
return data, stderr, proc.returncode
動作中：
In [2]: exec_command(['sleep', '10'], 5)
Out[2]: ('', '', -9)
In [3]: exec_command(['sleep', '10'], 11)
Out[3]: ('', '', 0)
Linuxコマンドの前に付けることtimeoutは悪い回避策ではなく、私にとってはうまくいきました。
cmd = "timeout 20 "+ cmd
subprocess.Popen(cmd.split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(output, err) = p.communicate()
私が使用する解決策は、シェルコマンドの前にtimelimitを付けることです。コマンドに時間がかかりすぎると、timelimitによって停止され、Popenにはtimelimitによって設定されたリターンコードが設定されます。128より大きい場合は、時間制限によってプロセスが強制終了されたことを意味します。
タイムアウトと大きな出力（> 64K）を使用したPythonサブプロセスも参照してください。
Python 2を使用している場合は、試してみてください
import subprocess32
try:
output = subprocess32.check_output(command, shell=True, timeout=3)
except subprocess32.TimeoutExpired as e:
print e
これらのいくつかから収集できるものを実装しました。これはWindowsで機能します。これはコミュニティウィキなので、コードも共有すると思います。
class Command(threading.Thread):
def __init__(self, cmd, outFile, errFile, timeout):
threading.Thread.__init__(self)
self.cmd = cmd
self.process = None
self.outFile = outFile
self.errFile = errFile
self.timed_out = False
self.timeout = timeout
def run(self):
self.process = subprocess.Popen(self.cmd, stdout = self.outFile, \
stderr = self.errFile)
while (self.process.poll() is None and self.timeout > 0):
time.sleep(1)
self.timeout -= 1
if not self.timeout > 0:
self.process.terminate()
self.timed_out = True
else:
self.timed_out = False
次に、別のクラスまたはファイルから：
outFile = tempfile.SpooledTemporaryFile()
errFile = tempfile.SpooledTemporaryFile()
executor = command.Command(c, outFile, errFile, timeout)
executor.daemon = True
executor.start()
executor.join()
if executor.timed_out:
out = 'timed out'
else:
outFile.seek(0)
errFile.seek(0)
out = outFile.read()
err = errFile.read()
outFile.close()
errFile.close()
* unixで完全なプロセスを実行する機械を理解すると、より簡単な解決策を簡単に見つけることができます。
select.select（）を使用してタイムアウト可能なcommunicate（）methを作成する方法のこの簡単な例を考えてみてください（最近は* nixでほとんどすべて利用可能です）。これはepoll/poll / kqueueで書くこともできますが、select.select（）バリアントが良い例かもしれません。また、select.select（）の主な制限（速度と1024 max fds）は、タスクには適用できません。
これは*nixで動作し、スレッドを作成せず、シグナルを使用せず、任意のスレッド（メインだけでなく）から起動でき、マシン（i5 2.3ghz）のstdoutから250mb/sのデータを読み取るのに十分な速度です。
通信の最後にstdout/stderrを結合する際に問題があります。巨大なプログラム出力がある場合、これは大きなメモリ使用量につながる可能性があります。ただし、より小さなタイムアウトで、communicate（）を数回呼び出すことができます。
class Popen(subprocess.Popen):
def communicate(self, input=None, timeout=None):
if timeout is None:
return subprocess.Popen.communicate(self, input)
if self.stdin:
# Flush stdio buffer, this might block if user
# has been writing to .stdin in an uncontrolled
# fashion.
self.stdin.flush()
if not input:
self.stdin.close()
read_set, write_set = [], []
stdout = stderr = None
if self.stdin and input:
write_set.append(self.stdin)
if self.stdout:
read_set.append(self.stdout)
stdout = []
if self.stderr:
read_set.append(self.stderr)
stderr = []
input_offset = 0
deadline = time.time() + timeout
while read_set or write_set:
try:
rlist, wlist, xlist = select.select(read_set, write_set, [], max(0, deadline - time.time()))
except select.error as ex:
if ex.args[0] == errno.EINTR:
continue
raise
if not (rlist or wlist):
# Just break if timeout
# Since we do not close stdout/stderr/stdin, we can call
# communicate() several times reading data by smaller pieces.
break
if self.stdin in wlist:
chunk = input[input_offset:input_offset + subprocess._PIPE_BUF]
try:
bytes_written = os.write(self.stdin.fileno(), chunk)
except OSError as ex:
if ex.errno == errno.EPIPE:
self.stdin.close()
write_set.remove(self.stdin)
else:
raise
else:
input_offset += bytes_written
if input_offset >= len(input):
self.stdin.close()
write_set.remove(self.stdin)
# Read stdout / stderr by 1024 bytes
for fn, tgt in (
(self.stdout, stdout),
(self.stderr, stderr),
):
if fn in rlist:
data = os.read(fn.fileno(), 1024)
if data == '':
fn.close()
read_set.remove(fn)
tgt.append(data)
if stdout is not None:
stdout = ''.join(stdout)
if stderr is not None:
stderr = ''.join(stderr)
return (stdout, stderr)
あなたはを使用してこれを行うことができますselect
import subprocess
from datetime import datetime
from select import select
def call_with_timeout(cmd, timeout):
started = datetime.now()
sp = subprocess.Popen(cmd, stdout=subprocess.PIPE)
while True:
p = select([sp.stdout], [], [], timeout)
if p[0]:
p[0][0].read()
ret = sp.poll()
if ret is not None:
return ret
if (datetime.now()-started).total_seconds() > timeout:
sp.kill()
return None
Python 2.7
import time
import subprocess
def run_command(cmd, timeout=0):
start_time = time.time()
df = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
while timeout and df.poll() == None:
if time.time()-start_time >= timeout:
df.kill()
return -1, ""
output = '\n'.join(df.communicate()).strip()
return df.returncode, output
Python 3.7.8でテストされたタイムアウト後にキャプチャされた出力の例：
try:
return subprocess.run(command, shell=True, capture_output=True, timeout=20, cwd=cwd, universal_newlines=True)
except subprocess.TimeoutExpired as e:
print(e.output.decode(encoding="utf-8", errors="ignore"))
assert False;
例外subprocess.TimeoutExpiredには、出力とその他のメンバーがあります。
cmd-子プロセスを生成するために使用されたコマンド。
タイムアウト-秒単位のタイムアウト。
output-run（）またはcheck_output（）によってキャプチャされた場合の子プロセスの出力。それ以外の場合は、なし。
stdout-stderrとの対称性のための出力のエイリアス。
stderr-run（）によってキャプチャされた場合の子プロセスのStderr出力。それ以外の場合は、なし。
詳細：https ：//docs.python.org/3/library/subprocess.html#subprocess.TimeoutExpired
私はWindows、Linux、Macでkillableprocessを正常に使用しました。Cygwin Pythonを使用している場合は、OSAFバージョンのkillableprocessが必要になります。そうしないと、ネイティブのWindowsプロセスが強制終了されないためです。
あまり詳しく調べていませんが、ActiveStateで見つけたこのデコレータは、この種のことには非常に役立つようです。と一緒subprocess.Popen(..., close_fds=True)に、少なくとも私はPythonでシェルスクリプトを作成する準備ができています。
このソリューションは、shell = Trueの場合にプロセスツリーを強制終了し、パラメーターをプロセスに渡し（または渡さず）、タイムアウトを設定し、コールバックのstdout、stderr、およびプロセス出力を取得します（kill_proc_treeにpsutilを使用します）。これは、jcolladoを含むSOに投稿されたいくつかのソリューションに基づいていました。jcolladoの回答におけるAnsonとjradiceのコメントに応じて投稿する。WindowsSrvr2012およびUbuntu14.04でテスト済み。Ubuntuの場合、parent.children（...）呼び出しをparent.get_children（...）に変更する必要があることに注意してください。
def kill_proc_tree(pid, including_parent=True):
parent = psutil.Process(pid)
children = parent.children(recursive=True)
for child in children:
child.kill()
psutil.wait_procs(children, timeout=5)
if including_parent:
parent.kill()
parent.wait(5)
def run_with_timeout(cmd, current_dir, cmd_parms, timeout):
def target():
process = subprocess.Popen(cmd, cwd=current_dir, shell=True, stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.PIPE)
# wait for the process to terminate
if (cmd_parms == ""):
out, err = process.communicate()
else:
out, err = process.communicate(cmd_parms)
errcode = process.returncode
thread = Thread(target=target)
thread.start()
thread.join(timeout)
if thread.is_alive():
me = os.getpid()
kill_proc_tree(me, including_parent=False)
thread.join()
Popenクラスをサブクラス化し、いくつかの単純なメソッドデコレータで拡張するというアイデアがあります。ExpirablePopenと呼びましょう。
from logging import error
from subprocess import Popen
from threading import Event
from threading import Thread
class ExpirablePopen(Popen):
def __init__(self, *args, **kwargs):
self.timeout = kwargs.pop('timeout', 0)
self.timer = None
self.done = Event()
Popen.__init__(self, *args, **kwargs)
def __tkill(self):
timeout = self.timeout
if not self.done.wait(timeout):
error('Terminating process {} by timeout of {} secs.'.format(self.pid, timeout))
self.kill()
def expirable(func):
def wrapper(self, *args, **kwargs):
# zero timeout means call of parent method
if self.timeout == 0:
return func(self, *args, **kwargs)
# if timer is None, need to start it
if self.timer is None:
self.timer = thr = Thread(target=self.__tkill)
thr.daemon = True
thr.start()
result = func(self, *args, **kwargs)
self.done.set()
return result
return wrapper
wait = expirable(Popen.wait)
communicate = expirable(Popen.communicate)
if __name__ == '__main__':
from subprocess import PIPE
print ExpirablePopen('ssh -T git@bitbucket.org', stdout=PIPE, timeout=1).communicate()
指定されたタイムアウト長より長くかかった場合、マルチスレッドサブプロセスを終了したいという問題がありました。でタイムアウトを設定したかったのですPopen()が、機能しませんでした。Popen().wait()次に、それが等しいことに気づいたので、メソッドcall()内にタイムアウトを設定することを考えました。これは最終的に機能しました。.wait(timeout=xxx)したがって、私はそれをこのように解決しました：
import os
import sys
import signal
import subprocess
from multiprocessing import Pool
cores_for_parallelization = 4
timeout_time = 15 # seconds
def main():
jobs = [...YOUR_JOB_LIST...]
with Pool(cores_for_parallelization) as p:
p.map(run_parallel_jobs, jobs)
def run_parallel_jobs(args):
# Define the arguments including the paths
initial_terminal_command = 'C:\\Python34\\python.exe' # Python executable
function_to_start = 'C:\\temp\\xyz.py' # The multithreading script
final_list = [initial_terminal_command, function_to_start]
final_list.extend(args)
# Start the subprocess and determine the process PID
subp = subprocess.Popen(final_list) # starts the process
pid = subp.pid
# Wait until the return code returns from the function by considering the timeout.
# If not, terminate the process.
try:
returncode = subp.wait(timeout=timeout_time) # should be zero if accomplished
except subprocess.TimeoutExpired:
# Distinguish between Linux and Windows and terminate the process if
# the timeout has been expired
if sys.platform == 'linux2':
os.kill(pid, signal.SIGTERM)
elif sys.platform == 'win32':
subp.terminate()
if __name__ == '__main__':
main()
遅い答えLinuxだけですが、誰かがsubprocess.getstatusoutput()タイムアウト引数を使用できない場合に使用したい場合は、コマンドの先頭に組み込みのLinuxタイムアウトを使用できます。
import subprocess
timeout = 25 # seconds
cmd = f"timeout --preserve-status --foreground {timeout} ping duckgo.com"
exit_c, out = subprocess.getstatusoutput(cmd)
if (exit_c == 0):
print("success")
else:
print("Error: ", out)
timeout引数：
--preserve-status：終了ステータスの保持--foreground：フォアグラウンドで実行25：秒単位のタイムアウト値残念ながら、私は雇用主によるソースコードの開示に関して非常に厳格なポリシーに拘束されているため、実際のコードを提供することはできません。しかし、私の好みでは、Popen.wait()無期限に待機するのではなく、ポーリングをオーバーライドするサブクラスを作成しPopen.__init__、タイムアウトパラメーターを受け入れるのが最善の解決策です。これを行うと、を含む他のすべてのPopenメソッド（を呼び出すwait）は期待どおりに機能しcommunicateます。
https://pypi.python.org/pypi/python-subprocess2は、サブプロセスモジュールの拡張機能を提供します。これにより、特定の期間まで待機できます。それ以外の場合は終了します。
したがって、プロセスが終了するまで最大10秒待機します。それ以外の場合は、強制終了します。
pipe = subprocess.Popen('...')
timeout = 10
results = pipe.waitOrTerminate(timeout)
これは、WindowsとUNIXの両方と互換性があります。「results」は辞書であり、アプリのリターンである「returnCode」（または、強制終了する必要がある場合は「None」）と「actionTaken」が含まれています。これは、プロセスが正常に完了した場合は「SUBPROCESS2_PROCESS_COMPLETED」になり、実行されたアクションに応じて「SUBPROCESS2_PROCESS_TERMINATED」とSUBPROCESS2_PROCESS_KILLEDのマスクになります（詳細については、ドキュメントを参照してください）。
Python 2.6以降の場合は、geventを使用します
from gevent.subprocess import Popen, PIPE, STDOUT
def call_sys(cmd, timeout):
p= Popen(cmd, shell=True, stdout=PIPE)
output, _ = p.communicate(timeout=timeout)
assert p.returncode == 0, p. returncode
return output
call_sys('./t.sh', 2)
# t.sh example
sleep 5
echo done
exit 1
もっと簡単なものを書こうとしていました。
#!/usr/bin/python
from subprocess import Popen, PIPE
import datetime
import time
popen = Popen(["/bin/sleep", "10"]);
pid = popen.pid
sttime = time.time();
waittime = 3
print "Start time %s"%(sttime)
while True:
popen.poll();
time.sleep(1)
rcode = popen.returncode
now = time.time();
if [ rcode is None ] and [ now > (sttime + waittime) ] :
print "Killing it now"
popen.kill()