timeout un sous-processus

Question

Je réalise qu'il pourrait s'agir d'un doublon de Using module 'subprocess' with timeout. Si c'est le cas, je m'excuse, je voulais juste clarifier quelque chose.

Je crée un sous-processus, que je veux exécuter pendant un certain temps, et s'il ne se termine pas dans ce délai, je veux qu'il génère une erreur. Est-ce que quelque chose dans le sens du code suivant fonctionnerait ou devons-nous utiliser un signal comme répondu dans l'autre question? Merci d'avance !:

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

Il serait, mais il a un problème. Le processus continuera à faire tout ce que vous lui demanderez de faire même après avoir abandonné. Vous devrez envoyer un signal au processus pour le tuer une fois que vous l'aurez abandonné si vous voulez vraiment qu'il s'arrête.

Puisque vous générez un nouveau processus (./configure qui est probablement un script configure) qui crée à son tour une tonne de sous-processus, cela va devenir un peu plus complexe.

import os 

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

Alors os.kill(-process.pid, signal.SIGKILL) devrait tuer tous les sous-processus. Fondamentalement ce que vous faites utilise le preexec_fn pour causer votre nouveau sous-processus à acquire it's own session group. Ensuite, vous envoyez un signal à tous les processus de ce groupe de sessions.

De nombreux processus qui génèrent des sous-processus savent qu'ils doivent nettoyer leurs sous-processus avant de mourir. Donc, il vous appartient d'essayer d'être gentil avec eux si vous le pouvez. Essayez d'abord os.signal(-process.pid, signal.SIGTERM), attendez une seconde ou deux pour que le processus se termine, puis essayez SIGKILL. Quelque chose comme ceci:

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

Comme une note de côté, jamais, jamais utiliser shell=True sauf si vous savez avec certitude absolue que est ce que vous voulez. Sérieusement. shell=True est carrément dangereux et la source de nombreux problèmes de sécurité et de comportement mystérieux.