Python3 Windows multiprocessing이 소켓을 처리하도록 처리합니다.

Question

Windows에서 멀티 프로세싱 ServerApp을 작동 시키려고합니다. 나는 os.fork() 기능이 없어서 socket을 피할 수없는 방법으로 전달해야 할 것입니다 (?!).

내가 here 표시하지만, 그 방법은 파이썬 3에서 사용할 수없는 것처럼이 multiprocessing.reduction에서 reduce_handle 및 rebuild_handle를 사용 할 수있을 것으로 보았다 (?!). 사용할 수있는 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에 등록합니다. 예를 들어 :

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의 산세에 unrealted 다른 문제가 나타났습니다.

• 대상으로 사용 self.conn_handler는 멀티 프로세싱은 전체 개체 self 피클을 시도합니다

  . 물체가 피클 할 수없는 일부 Thread을 포함하고 있기 때문에 문제가됩니다. 따라서 대상 함수가 닫히면 self을 제거해야합니다. @staticmethod 데코레이터를 사용하고 기능에서 self의 모든 언급을 제거하여 수행 할 수 있습니다.

• 또한 logging 모듈은 여러 프로세스를 처리하지 않습니다. 기본적으로 시작된 Process의 모든 로그는 현재 코드로 손실됩니다. 이 문제를 해결하려면 두 번째 Process (시작 부분이 conn_handler)을 시작하거나 multiprocessing 로깅 유틸리티를 사용하면 새로운 logging을 시작할 수 있습니다.

이 캔은 다음과 같이 제공합니다

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() 

난 단지 유닉스와 python3.6에 테스트를하지만 난 창에서 스폰 컨텍스트 , which should behave like the Process`를 사용할 때 너무 다른 행동을하지 말았어야 .