시간 초과와 함께 모듈 '하위 프로세스'사용

다음은 stdout데이터를 반환하는 임의의 명령을 실행 하거나 0이 아닌 종료 코드에서 예외를 발생 시키는 Python 코드입니다.

proc = subprocess.Popen(
    cmd,
    stderr=subprocess.STDOUT,  # Merge stdout and stderr
    stdout=subprocess.PIPE,
    shell=True)

communicate 프로세스가 종료되기를 기다리는 데 사용됩니다.

stdoutdata, stderrdata = proc.communicate()

이 subprocess모듈은 시간 초과 (X 초 이상 실행중인 프로세스를 종료하는 기능)를 지원하지 않으므로 실행하는 데 시간이 오래 communicate걸릴 수 있습니다.

Windows 및 Linux에서 실행되는 Python 프로그램에서 시간 초과를 구현 하는 가장 간단한 방법 은 무엇입니까 ?

답변

파이썬 3.3 이상에서 :

from subprocess import STDOUT, check_output

output = check_output(cmd, stderr=STDOUT, timeout=seconds)

output 명령의 병합 된 stdout, stderr 데이터를 포함하는 바이트 문자열입니다.

check_output인상 CalledProcessError과는 달리 문제의 텍스트에 지정된대로 0이 아닌 종료 상태에 대한 proc.communicate()방법.

shell=True불필요하게 자주 사용되기 때문에 제거 했습니다. cmd실제로 필요한 경우 언제든지 다시 추가 할 수 있습니다 . 당신이 추가 할 경우 shell=True즉, 자식 프로세스의 급부상 경우 자신의 후손; check_output()제한 시간이 표시 한 것보다 훨씬 늦게 리턴 될 수 있습니다 ( 서브 프로세스 제한 시간 실패 참조) .

타임 아웃 기능은 subprocess323.2+ 서브 프로세스 모듈 의 백 포트를 통해 Python 2.x에서 사용할 수 있습니다 .

답변

나는 낮은 수준의 세부 사항에 대해 많이 모른다. 그러나 파이썬 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) 두 번째 프로세스에서 프로세스가 종료되었음을 알 수 있습니다 (리턴 코드 -15).

나는 창문에서 테스트하지 않았습니다. 그러나 예제 명령을 업데이트하는 것 외에도 thread.join 또는 process.terminate가 지원되지 않는다는 것을 문서에서 찾지 못했기 때문에 작동해야한다고 생각합니다.

답변

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

답변

유닉스에 있다면

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

답변

다음은 적절한 프로세스 종료 기능이있는 모듈로서 Alex Martelli의 솔루션입니다. 다른 접근 방식은 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)

답변

sussudio 답변을 수정 했습니다 . 이제 수익을 기능 : ( returncode, stdout, stderr, timeout) – stdout와 stderrUTF-8 문자열로 디코딩

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"]

답변

아무도 사용을 언급하지 않았다 timeout

timeout 5 ping -c 3 somehost

이것은 모든 유스 케이스에 분명히 적용되는 것은 아니지만 간단한 스크립트를 다루는 경우 이길 수 없습니다.

homebrewmac 사용자 를 위해 coreutils에서 gtimeout으로도 제공됩니다 .

Exit mobile version

How IT

언제든지 물어보세요.

시간 초과와 함께 모듈 ‘하위 프로세스’사용

답변

답변

답변

답변

답변

답변

답변