Python中的多处理和套接字

Question

我试图使多处理和socket编程一起工作，但是，我在这一点上陷入了困境。问题是，我得到了这个错误：

  File "multiprocesssockserv.py", line 11, in worker
    clientsocket = socket.fromfd(clientfileno, socket.AF_INET, socket.SOCK_STREAM)
error: [Errno 9] Bad file descriptor

导致错误的完整代码如下：

import multiprocessing as mp
import logging
import socket

logger = mp.log_to_stderr(logging.WARN)

def worker(queue):
    while True:
        clientfileno = queue.get()
        print clientfileno
        clientsocket = socket.fromfd(clientfileno, socket.AF_INET, socket.SOCK_STREAM)
        clientsocket.recv()
        clientsocket.send("Hello World")
        clientsocket.close()

if __name__ == '__main__':
    num_workers = 5
    socket_queue = mp.Queue()
    workers = [mp.Process(target=worker, args=(socket_queue,)) for i in
            range(num_workers)]

    for p in workers:
        p.daemon = True
        p.start()

    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serversocket.bind(('',9090))
    serversocket.listen(5)
    while True:
        client, address = serversocket.accept()
        socket_queue.put(client.fileno())

编辑：我正在使用socket.fromfd，因为我无法将套接字放入队列：）我需要一种方法以某种方式从不同进程访问相同的套接字。这是我问题的核心。

Answer 1

我不是专家，所以我不能给出真正的解释，但是如果您想使用队列，您需要减少句柄，然后重新创建它：

在您的main中：

client, address = serversocket.accept()
client_handle = multiprocessing.reduction.reduce_handle(client.fileno())
socket_queue.put(client_handle)

在你的工人身上：

clientHandle = queue.get()
file_descriptor = multiprocessing.reduction.rebuild_handle(client_handle)
clientsocket = socket.fromfd(file_descriptor, socket.AF_INET, socket.SOCK_STREAM)

同时

import multiprocessing.reduction

这将适用于您的原始代码。 但是，在按照我描述的方式创建工作进程之后，我目前在关闭这些进程中的套接字时遇到问题。

Answer 2

下面是上面提到的一些工作代码-https://gist.github.com/sunilmallya/4662837多处理。还原套接字服务器的父进程在接受连接后将连接传递给客户端

Answer 3

经过一段时间的研究，我决定从另一个角度来解决这个问题，下面的方法似乎对我很有用。

import multiprocessing as mp
import logging
import socket
import time

logger = mp.log_to_stderr(logging.DEBUG)

def worker(socket):
    while True:
        client, address = socket.accept()
        logger.debug("{u} connected".format(u=address))
        client.send("OK")
        client.close()
if __name__ == '__main__':
    num_workers = 5

    serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    serversocket.bind(('',9090))
    serversocket.listen(5)

    workers = [mp.Process(target=worker, args=(serversocket,)) for i in
            range(num_workers)]

    for p in workers:
        p.daemon = True
        p.start()

    while True:
        try:
            time.sleep(10)
        except:
            break