时间戳

首页 > 解决方案 > 杀死进程，包含一个无限循环，显然在多个操作系统上

python - 杀死进程，包含一个无限循环，显然在多个操作系统上

问题描述

我有一个 Python3 程序，应该在 linux 和 windows 上用 Python 3.6+ 进行测试。

因为 testenviroment 是一个需要的 TestServer，它应该在后台作为进程（来自多处理）运行，而测试正在运行。测试与单元测试一起运行。

TestServer 进程（称为 mainloop）如下所示：

def mainloop(somearg):
    server = mylib.server.Server()
    # some initialisation stuff
    server.start()
    while True: # How to break/handle this if process shall be killed?
        while True:  # How to break/handle this if process shall be killed?
            event = server.get_event()
            if event:
                logger.info(event)
            else:
                break
        time.sleep(1)
    server.stop() # never been reached
    server.cleanup()  # never been reached

我的 setUpClass 看起来像这样：

from multiprocessing import Process
from mylib import mainloop

# some other stuff

@classmethod
def setUpClass(cls):
    cls.process = Process(target=mainloop)
    cls.process.start()
    time.sleep(2) # wait for server to start

@classmethod
def tearDownClass(cls):
    kill(cls.process.pid,1)

问题：

  1. 如果“mainloop”应该被杀死，我该如何处理，它能够停止并以适当的方式清理？
  2. 我如何为 Windows 和 linux 系统管理它？
  3. 这种测试目的的最佳实践是什么（处理多个操作系统并在每个多个 python 版本上进行测试）？

标签： pythonpython-3.xoperating-systemmultiprocessingpython-unittest

解决方案

基本上有两种主要方法可以正确关闭服务器。

  1. 向服务器发送一条特殊消息，告诉他他应该关闭
  2. 杀死服务器运行的进程

似乎您的代码使用了后者，而不是专注于这一点。os.kill您确实通过使用很棒的功能发送了一个停止服务器进程的信号！

但是你需要对你的代码做一些调整，让它像你想要的那样工作。首先让我们定义想要什么：

  1. 函数tearDownClass将向服务器发送信号以停止
  2. 服务器将收到该信号并优雅退出

让我们进行调整：
1。

import signal  # built-in library for signals
import os
...

@classmethod
def tearDownClass(cls):
    # this is very uncomfortable but windows makes it hard because they have
    # their own signals
    send_signal = signal.SIGINT if os.name == "posix" else signal.CTRL_C_EVENT
    kill(cls.process.pid,send_signal)
  1. 现在我们需要处理当信号到达进程时会发生什么
def mainloop(somearg):
    server = mylib.server.Server()
    # some initialisation stuff
    server.start()

    # defining a callback for cleanup
    def gracfull_cleanup():
        server.stop()
        server.cleanup()

    # again registering the proper signals to the OS
    receiving_signal = signal.SIGINT if os.name == "posix" else signal.CTRL_C_EVENT
    signal.signal(receiving_signal, gracfull_cleanup)

    while True: # How to break/handle this if process shall be killed?
        while True:  # How to break/handle this if process shall be killed?
            event = server.get_event()
            if event:
                logger.info(event)
            else:
                break
        time.sleep(1)
    server.stop() # never been reached
    server.cleanup()  # never been reached

这应该可以解决问题。请注意，现在一旦您发送信号，您的处理程序将开始工作并为您进行清理。

最后一点！ 我没有使用 Windows，所以我无法测试它，如果信号不起作用，请尝试 SIGILL

推荐阅读