python - 拥有子进程允许 rpc-server 在子进程存活时重新启动

Question

场景

我有一个 rpc-server 需要生成持续数天的重要进程 (multiprocessing.Process)。出于安全原因，我不希望这些进程的生存依赖于 rpc-server。因此，我希望服务器能够在进程运行时并能够重新启动。

孤立进程

这个问题可以通过以下方式解决(不要将它粘贴到您不想丢失之前工作的地方，它会关闭您的 python session ):

import os
import multiprocessing
import time

def _job(data):
    for _ in range(3):
        print multiprocessing.current_process(), "is working"
        time.sleep(2)
    print multiprocessing.current_process(), "is done"

#My real worker gets a Connection-object as part of a
#multiprocessing.Pipe among other arguments
worker = multiprocessing.Process(target=_job, args=(None,))
worker.daemon = True
worker.start()
os._exit(0)

问题:如果 worker 处于事件状态，则关闭 rpc-server 的套接字

退出主进程似乎无助于或影响套接字问题的关闭。因此，为了说明服务器重新启动的问题，模拟了在第一个服务器关闭后启动具有相同参数的第二个服务器。

以下完美运行:

import SimpleXMLRPCServer
HOST = "127.0.0.1"
PORT = 45212
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
s.server_close()
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
s.server_close()

但是，如果创建了一个 worker，它会引发一个 socket.error 说明套接字已在使用中:

s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
worker = multiprocessing.Process(target=_job, args=(None,))
worker.start()
s.server_close()
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT)) #raises socket.error
worker.join()
s.server_close()

手动关闭服务器套接字确实有效:

import socket
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
worker = multiprocessing.Process(target=_job, args=(None,))
worker.start()
s.socket.shutdown(socket.SHUT_RDWR)
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
worker.join()
s.server_close()

但是这种行为真的让我很担心。我没有以任何方式将套接字传递给工作人员，但它似乎无论如何都捕获了它。

以前也有类似的问题，但它们倾向于将套接字传递给工作人员，这不是这里的目的。如果我通过发送套接字，我可以在 worker 中关闭它并绕过 shutdown hack:

def _job2(notMySocket):
    notMySocket.close()
    for _ in range(3):
        print multiprocessing.current_process(), "is working"
        time.sleep(2)
    print multiprocessing.current_process(), "is done"

s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
worker = multiprocessing.Process(target=_job2, args=(s.socket,))
worker.start()
time.sleep(0.1) #Just to be sure worker gets to close socket in time
s.server_close()
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT)) 
worker.join()
s.server_close()

但是服务器的socket绝对没有理由去访问worker。我有点不喜欢这个解决方案，即使它是迄今为止最好的解决方案。

问题

有没有一种方法可以限制在使用 multiprocessing.Process 时 fork 的内容，以便只复制我想传递给目标的内容，而不是所有打开的套接字和其他内容？

就我而言，要使此代码正常工作:

s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT))
childPipe, parentPipe = multiprocessing.Pipe()
worker = multiprocessing.Process(target=_job, args=(childPipe,))
worker.start()
s.server_close()
s = SimpleXMLRPCServer.SimpleXMLRPCServer((HOST, PORT)) #raises socket.error
worker.join()
s.server_close()

Answer 1

如果您使用的是 Python 2.x，我认为没有任何方法可以避免 Posix 平台上的这种继承。 os.fork 将始终用于创建新进程，这意味着父进程的整个状态将被复制到子进程。您所能做的就是立即关闭 child 的套接字，这就是您已经在做的事情。避免这种继承的唯一方法是在启动服务器之前启动进程。您可以通过尽早启动 Process 然后使用 multiprocessing.Queue 来交付工作项(而不是 args 关键字)来做到这一点argument) 或 multiprocessing.Event 来指示它应该实际开始工作。根据您需要发送给子进程的内容，您的用例可能实际可行，也可能不可行。

但是，如果您使用的是 Python 3.4+(或可以迁移到 3.4+)，则可以使用 spawn or forkserver contexts以避免继承套接字。

spawn

The parent process starts a fresh python interpreter process. The child process will only inherit those resources necessary to run the process objects run() method. In particular, unnecessary file descriptors and handles from the parent process will not be inherited. Starting a process using this method is rather slow compared to using fork or forkserver.

Available on Unix and Windows. The default on Windows.

forkserver

When the program starts and selects the forkserver start method, a server process is started. From then on, whenever a new process is needed, the parent process connects to the server and requests that it fork a new process. The fork server process is single threaded so it is safe for it to use os.fork(). No unnecessary resources are inherited.

例子:

def _job2():
    for _ in range(3):
        print multiprocessing.current_process(), "is working"
        time.sleep(2)
    print multiprocessing.current_process(), "is done"

ctx = multiprocessing.get_context('forkserver')
worker = ctx.Process(target=_job2)
worker.start()