具有超时的异步子进程

Question

我在Python 3中生成具有超时的异步子进程时遇到问题.

我想要实现的目标:我希望异步生成多个进程而不等待结果,但我还要确保每个生成的进程都会在给定的超时内结束.

我在这里发现了类似的问题:在Python中使用模块'subprocess'和超时以及异步后台进程？但他们没有解决我的问题.

我的代码看起来像这样.我有使用模块'subprocess'和超时建议的Command类:

class Command(object):
  def __init__(self, cmd):
    self.cmd = cmd
    self.process = None

  def run(self, timeout):
    def target():
      print('Thread started')
      args = shlex.split(self.cmd)
      self.process = subprocess.Popen(args, 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()

然后当我想要产生子进程时:

for system in systems:
  for service in to_spawn_system_info:
    command_str = "cd {0} && python proc_ip.py {1} {2} 0 2>>{3}".format(home_dir,
        service, system, service_log_dir)
    command = Command(command_str)
    command.run(timeout=60)

当我运行它时,输出似乎等待每个命令生成并结束.我明白了

Thread started
Thread finished
Thread started
Thread finished
Thread started
Thread finished
Thread started
Thread finished

所以我的问题是我做错了什么？现在我开始怀疑是否有可能产生进程并通过超时限制其执行.

为什么我需要这个？spawner脚本将在cron中运行.它将每10分钟执行一次,它必须产生大约20个子进程.我想保证每个子进程都会在脚本再次从cron运行之前结束.

Answer 1

如前所述，对 process.communicate() 的调用使您的代码等待子流程完成。但是，如果您只是删除 communications() 调用，线程将在生成进程后立即退出，导致 thread.join() 调用过早退出，并且您将过早终止子进程。要在不轮询或忙等待的情况下执行您想要的操作，您可以设置一个计时器，如果进程尚未完成，该计时器将在超时后终止进程（和运行程序线程）：

class Command(object):
  def __init__(self, cmd):
    self.cmd = cmd
    self.process = None

  def run(self, timeout):
    def target():
      print('Thread started')
      # May want/need to skip the shlex.split() when using shell=True
      # See Popen() constructor docs on 'shell' argument for more detail.
      args = shlex.split(self.cmd)
      self.process = subprocess.Popen(args, shell=True)
      self.timer.start()
      self.process.wait()
      self.timer.cancel()

    def timer_callback():
        print('Terminating process (timed out)')
        self.process.terminate()

    thread = threading.Thread(target=target)
    self.timer = threading.Timer(timeout, timer_callback)
    thread.start()