Kill Process，包含一个无限循环，显然在多个 os

Question

我有一个 Python3 程序，应使用 linux 和 windows 以及 Python 3.6+ 进行测试。

因为 os 测试环境是一个需要的测试服务器，它应该运行在后台作为进程（来自多处理），而测试是运行宁。测试是运行个单元测试。

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)

问题：

如果“mainloop”被终止，我该如何处理它能够停止并以正确的方式清理？
我如何管理 windows 和 linux 系统？
此类测试 purposes 的最佳实践是什么（处理多个 os 并在每个多个 python 版本上进行测试）？

Answer 1

非常简单，就像您可以使用 kill 终止您的服务器进程一样。

您可以向它发送信号。如果您的服务器进程已经注册了一个信号处理程序。然后就可以正常退出了。

但这只适用于Unix-like系统（某些信号在Windows中无法实现）。如果你想让你的程序在 Windows/Unix-like 系统上运行，你可以将 signal 机制扩展到更通用的 RPC.

例如，用TCP套接字监听一个端口。当你想从外面关闭时，给它发消息。

Answer 2

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

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

您的代码似乎使用了后者，我将重点关注它。您确实使用 os.kill 函数发送了一个停止服务器进程的信号，这很棒！

但是您需要对您的代码进行一些调整，以使其按您希望的方式运行。首先让我们定义想要什么：

函数tearDownClass会向服务器发送一个信号停止
服务器将收到该信号并正常退出

让我们进行调整：
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)

现在我们需要处理当信号到达进程时发生的事情

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

Kill Process，包含一个无限循环，显然在多个 os

Kill Process, containing an infinite loop, clearly on multiple os

python

operating-system

multiprocessing

python-3.x

python-unittest