таймаут подпроцесс

Question

Я понимаю, что это может быть дубликат Using module 'subprocess' with timeout. Если это так, я извиняюсь, просто хотел прояснить что-то.

Я создаю подпроцесс, который я хочу запустить в течение определенного промежутка времени, и если он не завершится в течение этого времени, я хочу, чтобы он выдал ошибку. Будет ли что-то в соответствии с приведенным ниже кодом работать или нужно использовать сигнал, подобный ответу на другой вопрос? Заранее благодарю !:

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

Будет, но проблема возникла. Процесс будет продолжаться, делая то, что вы просили, чтобы он сделал это даже после того, как вы отказались от него. Вам нужно будет отправить процессу сигнал, чтобы убить его, как только вы отказались от него, если вы действительно хотите его остановить.

Поскольку вы создаете новый процесс (./configure, который предположительно является скриптом configure), который, в свою очередь, создает целую тонну подпроцессов, это будет немного сложнее.

import os 

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

Тогда os.kill(-process.pid, signal.SIGKILL) должен убить все подпроцессы. В основном, вы используете preexec_fn, чтобы ваш новый подпроцесс был acquire it's own session group. Затем вы отправляете сигнал всем процессам в этой группе сеансов.

Многие процессы, которые порождают подпроцессы, знают, что им необходимо очистить свои подпроцессы до их смерти. Так что вам должно быть приятно, если сможете. Сначала попробуйте os.signal(-process.pid, signal.SIGTERM), подождите секунду или два, чтобы завершить процесс, затем попробуйте SIGKILL. Что-то вроде этого:

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

В качестве примечания, никогда, никогда не использовать shell=True, если вы не знаете, для абсолютного уверен, что это то, что вы хотите. Шутки в сторону. shell=True является абсолютно опасным и источником многих проблем безопасности и таинственного поведения.