Python3 Windows 多处理将套接字传递给进程

Question

我正在尝试使多处理 ServerApp 在 Windows 上工作。我猜这个问题缺少os.fork() 功能，所以我必须以某种方式通过socket，这是不可腌制的（？！）。

我已经看到使用multiprocessing.reduction 中的reduce_handle 和rebuild_handle 可能会实现这一点，如here 所示，但这些方法在Python 3 中不可用（？！）。虽然我有可用的 duplicate 和 steal_handle 可用，但我找不到如何使用它们或我是否需要它们的示例。

另外，我想知道logging 在创建新进程时是否会成为问题？

这是我的 ServerApp 示例：

import logging
import socket

from select import select
from threading import Thread
from multiprocessing import Queue
from multiprocessing import Process
from sys import stdout
from time import sleep


class ServerApp(object):

    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler(stdout)
    formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)


    def conn_handler(self, connection, address, buffer):

        self.logger.info("[%d] - Connection from %s:%d", self.id, address[0], address[1])

        try:
            while True:

                command = None
                received_data = b''
                readable, writable, exceptional = select([connection], [], [], 0)  # Check for client commands

                if readable:
                    # Get Command  ... There is more code here
                    command = 'Something'


                if command == 'Something':
                    connection.sendall(command_response)
                else:
                    print(':(')

        except Exception as e:
            print(e)
        finally:
            connection.close()
            self.client_buffers.remove(buffer)
            self.logger.info("[%d] - Connection from %s:%d has been closed.", self.id, address[0], address[1])


    def join(self):

        while self.listener.is_alive():
            self.listener.join(0.5)


    def acceptor(self):

        while True:
            self.logger.info("[%d] - Waiting for connection on %s:%d", self.id, self.ip, self.port)

            # Accept a connection on the bound socket and fork a child process to handle it.
            conn, address = self.socket.accept()

            # Create Queue which will represent buffer for specific client and add it o list of all client buffers
            buffer = Queue()
            self.client_buffers.append(buffer)

            process = Process(target=self.conn_handler, args=(conn, address, buffer))
            process.daemon = True
            process.start()
            self.clients.append(process)

            # Close the connection fd in the parent, since the child process has its own reference.
            conn.close()


    def __init__(self, id, port=4545, ip='127.0.0.1', method='tcp', buffer_size=2048):

        self.id = id
        self.port = port
        self.ip = ip

        self.socket = None
        self.listener = None
        self.buffer_size = buffer_size

        # Additional attributes here....

        self.clients = []
        self.client_buffers = []


    def run(self):

        # Create TCP socket, bind port and listen for incoming connections
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.bind((self.ip, self.port))
        self.socket.listen(5)

        self.listener = Thread(target=self.acceptor)  # Run acceptor thread to handle new connection
        self.listener.daemon = True
        self.listener.start()

Answer 1

要允许 python3 的连接酸洗（包括套接字），您应该使用mulitprocessing.allow_connection_pickling。它为ForkingPickler 中的套接字注册reducer。例如：

import socket
import multiprocessing as mp
mp.allow_connection_pickling()


def _test_connection(conn):
    msg = conn.recv(2)
    conn.send(msg)
    conn.close()
    print("ok")

if __name__ == '__main__':
    server, client = socket.socketpair()

    p = mp.Process(target=_test_connection, args=(server,))
    p.start()

    client.settimeout(5)

    msg = b'42'
    client.send(msg)
    assert client.recv(2) == msg

    p.join()
    assert p.exitcode == 0

    client.close()
    server.close()

我还注意到您还有一些其他问题与socket 的酸洗无关。

• 当使用self.conn_handler 作为目标时，多处理将尝试腌制整个对象self。这是一个问题，因为您的对象包含一些无法腌制的 Thread。因此，您应该从目标函数的闭包中删除 self。可以通过使用 @staticmethod 装饰器并删除函数中所有提及 self 来完成。

• 另外，logging 模块不是用来处理多个进程的。基本上，来自已启动Process 的所有日志都将随着您当前的代码丢失。要解决此问题，您可以在启动第二个 Process（在 conn_handler 的开头）后启动一个新的 logging，或使用 multiprocessing 日志记录实用程序。

这可以给出这样的结果：

import logging
import socket

from select import select
from threading import Thread
from multiprocessing import util, get_context
from sys import stdout
from time import sleep

util.log_to_stderr(20)
ctx = get_context("spawn")


class ServerApp(object):

    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)
    handler = logging.StreamHandler(stdout)
    formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    def __init__(self, id, port=4545, ip='127.0.0.1', method='tcp',
                buffer_size=2048):

        self.id = id
        self.port = port
        self.ip = ip

        self.socket = None
        self.listener = None
        self.buffer_size = buffer_size

        # Additional attributes here....

        self.clients = []
        self.client_buffers = []

    @staticmethod
    def conn_handler(id, connection, address, buffer):

        print("test")
        util.info("[%d] - Connection from %s:%d", id, address[0], address[1])

        try:
            while True:

                command = None
                received_data = b''
                # Check for client commands
                readable, writable, exceptional = select([connection], [], [],
                                                        0)

                if readable:
                    # Get Command  ... There is more code here
                    command = 'Something'

                if command == 'Something':
                    connection.sendall(b"Coucouc")
                    break
                else:
                    print(':(')
                sleep(.1)

        except Exception as e:
            print(e)
        finally:
            connection.close()
            util.info("[%d] - Connection from %s:%d has been closed.", id,
                    address[0], address[1])
            print("Close")

    def join(self):

        while self.listener.is_alive():
            self.listener.join(0.5)

    def acceptor(self):

        while True:
            self.logger.info("[%d] - Waiting for connection on %s:%d", self.id,
                            self.ip, self.port)

            # Accept a connection on the bound socket and fork a child process
            # to handle it.
            conn, address = self.socket.accept()

            # Create Queue which will represent buffer for specific client and
            # add it o list of all client buffers
            buffer = ctx.Queue()
            self.client_buffers.append(buffer)

            process = ctx.Process(target=self.conn_handler,
                                args=(self.id, conn, address, buffer))
            process.daemon = True
            process.start()
            self.clients.append(process)

            # Close the connection fd in the parent, since the child process
            # has its own reference.
            conn.close()

    def run(self):

        # Create TCP socket, bind port and listen for incoming connections
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket.bind((self.ip, self.port))
        self.socket.listen(5)

        # Run acceptor thread to handle new connection
        self.listener = Thread(target=self.acceptor)
        self.listener.daemon = True
        self.listener.start()

        self.listener.join()


def main():
    app = ServerApp(0)
    app.run()


if __name__ == '__main__':
    main()

我只在 Unix 和 python3.6 上测试过它，但它的行为应该不会有太大的不同，因为我在 windows 中使用 spawn context, which should behave like theProcess`。