python - 具有多处理工作人员的扭曲网络客户端？

Question

所以，我有一个使用 Twisted + Stomper 作为 STOMP 客户端的应用程序，它将工作外包给多进程。工作人员池。

当我只使用 python 脚本启动它时，这似乎工作正常，(简化)看起来像这样:

# stompclient.py

logging.config.fileConfig(config_path)
logger = logging.getLogger(__name__)

# Add observer to make Twisted log via python
twisted.python.log.PythonLoggingObserver().start() 

# initialize the process pool.  (child processes get forked off immediately)
pool = multiprocessing.Pool(processes=processes)

StompClientFactory.username = username
StompClientFactory.password = password
StompClientFactory.destination = destination
reactor.connectTCP(host, port, StompClientFactory())
reactor.run()

当它被打包用于部署时，我想我会利用 twistd 脚本并从 tac 文件运行它。

这是我的 tac 文件:

# stompclient.tac

logging.config.fileConfig(config_path)
logger = logging.getLogger(__name__)

# Add observer to make Twisted log via python
twisted.python.log.PythonLoggingObserver().start() 

# initialize the process pool.  (child processes get forked off immediately)
pool = multiprocessing.Pool(processes=processes)

StompClientFactory.username = username
StompClientFactory.password = password
StompClientFactory.destination = destination

application = service.Application('myapp')

service = internet.TCPClient(host, port, StompClientFactory())
service.setServiceParent(application)

为了便于说明，我折叠或更改了一些细节；希望它们不是问题的本质。例如，我的应用程序有一个插件系统，池由一个单独的方法初始化，然后使用 pool.apply_async() 传递我的插件的 process() 方法之一将工作委托(delegate)给池。

因此，如果我运行脚本 (stompclient.py)，一切都会按预期进行。

如果我在非守护进程模式 (-n) 下运行 twist，它似乎也能正常工作:

twistd -noy stompclient.tac

但是，当我在守护进程模式下运行时，它不工作:

twistd -oy stompclient.tac

该应用程序似乎启动正常，但当它试图停止工作时，它只是挂起。 “挂起”是指子进程似乎从未被要求做任何事情，而父进程(称为 pool.apply_async())只是坐在那里等待响应返回。

我确定我在使用 Twisted + 多处理做一些愚蠢的事情，但我真的希望有人能向我解释我的方法中的缺陷。

提前致谢!

Answer 1

由于您的工作调用和非工作调用之间的区别仅在于“-n”选项，因此问题很可能是由守护进程引起的(“-n”可防止发生)。

在 POSIX 上，守护进程涉及的步骤之一是 fork 并让父进程退出。其中之一是，这会导致您的代码在与评估 .tac 文件的进程不同的进程中运行。这也重新安排了在 .tac 文件中启动的进程的子/父关系 - 正如您的多处理进程池一样。

多处理池的进程从您启动的 twistd 进程的父进程开始。但是，当该进程作为守护进程的一部分退出时，它们的父进程将成为系统初始化进程。这可能会导致一些问题，尽管可能不是您描述的挂起问题。可能还有其他类似的低级实现细节，这些细节通常允许多处理模块工作，但会被守护进程中断。

幸运的是，避免这种奇怪的交互应该很简单。 Twisted 的服务 API 允许您在守护进程完成后运行代码。如果您使用这些 API，那么您可以将多处理模块的进程池的初始化延迟到守护进程之后，并有望避免该问题。这是一个可能看起来像的示例:

from twisted.application.service import Service

class MultiprocessingService(Service):
    def startService(self):
        self.pool = multiprocessing.Pool(processes=processes)

MultiprocessingService().setServiceParent(application)

现在，单独地，您可能还会遇到与清理多处理模块的子进程相关的问题，或者可能遇到与使用 Twisted 的进程创建 API reactor.spawnProcess 创建的进程相关的问题。这是因为正确处理子进程的一部分通常涉及处理 SIGCHLD 信号。不过，Twisted 和 multiprocessing 不会在这方面进行合作，因此其中一个会收到所有子进程退出的通知，而另一个永远不会收到通知。如果您根本不使用 Twisted 的 API 来创建子进程，那么这对您来说可能没问题 - 但您可能需要检查以确保多处理模块尝试安装的任何信号处理程序实际上“获胜”并且没有得到替换为 Twisted 自己的处理程序。