Timeout ein Subprozess

Question

Ich weiß, dass dies ein Duplikat von Using module 'subprocess' with timeout sein könnte. Wenn es so ist, entschuldige ich mich, wollte nur etwas klären.

Ich erstelle einen Subprozess, den ich für eine bestimmte Zeit ausführen möchte, und wenn es nicht innerhalb dieser Zeit abgeschlossen ist, möchte ich, dass es einen Fehler auslöst. Würde etwas in der Form des folgenden Codes funktionieren oder müssen wir ein Signal verwenden, wie es in der anderen Frage beantwortet wurde? Vielen Dank im Voraus !:

def run(self): 
    self.runTestCmd() 
    self.waitTestComplete(self.timeout) 

def runTestCmd(self): 
    self.proc = subprocess.Popen("./configure", shell=True) 

def waitTestComplete(self, timeout): 
    st = time.time() 
    while (time.time()-st) < timeout: 
     if self.proc.poll() == 0: 
      return True 
     else: 
      time.sleep(2) 
    raise TestError("timed out waiting for test to complete") 

Answer 1

Es wäre, aber es hat ein Problem. Der Prozess wird weiterhin tun, was immer Sie wollten, auch nachdem Sie es aufgegeben haben. Sie müssen dem Prozess ein Signal senden, um es zu töten, sobald Sie es aufgegeben haben, wenn Sie wirklich wollen, dass es aufhört.

Da Sie einen neuen Prozess erzeugen (./configure, der vermutlich ein Konfigurationsskript ist), der wiederum eine ganze Tonne von Teilprozessen erzeugt, wird dies ein wenig komplexer werden.

import os 

def runTestCmd(self): 
    self.proc = subprocess.Popen(["./configure"], shell=False, 
           preexec_fn=os.setsid) 

Dann os.kill(-process.pid, signal.SIGKILL) sollten alle Teilprozesse töten. Grundsätzlich verwenden Sie die preexec_fn, um Ihren neuen Unterprozess zu acquire it's own session group zu verursachen. Dann senden Sie ein Signal an alle Prozesse in dieser Session-Gruppe.

Viele Prozesse, die Subprozesse generieren, wissen, dass sie ihre Subprozesse aufräumen müssen, bevor sie absterben. Also solltest du versuchen, nett zu ihnen zu sein, wenn du kannst. Versuchen Sie zuerst os.signal(-process.pid, signal.SIGTERM), warten Sie ein oder zwei Sekunden, bis der Prozess beendet ist, und versuchen Sie dann SIGKILL. Etwas wie folgt aus:

import time, os, errno, signal 

def waitTestComplete(self, timeout): 
    st = time.time() 
    while (time.time()-st) < timeout: 
     if self.proc.poll() is not None: # 0 just means successful exit 
      # Only return True if process exited successfully, 
      # otherwise return False. 
      return self.proc.returncode == 0 
     else: 
      time.sleep(2) 
    # The process may exit between the time we check and the 
    # time we send the signal. 
    try: 
     os.kill(-self.proc.pid, signal.SIGTERM) 
    except OSError, e: 
     if e.errno != errno.ESRCH: 
      # If it's not because the process no longer exists, 
      # something weird is wrong. 
      raise 
    time.sleep(1) 
    if self.proc.poll() is None: # Still hasn't exited. 
     try: 
      os.kill(-self.proc.pid, signal.SIGKILL) 
     except OSError, e: 
      if e.errno != errno.ESRCH: 
       raise 
    raise TestError("timed out waiting for test to complete") 

Als Randbemerkung, nie, nie shell=True verwenden, wenn Sie für absolute Sicherheit wissen, dass das, was Sie wollen. Ernst. shell=True ist geradezu gefährlich und die Quelle vieler Sicherheitsprobleme und mysteriösen Verhaltens.