<s> __all__ = [ ] <EOL> import os <EOL> import sys <EOL> import socket <EOL> import threading <EOL> import _multiprocessing <EOL> from multiprocessing import current_process <EOL> from multiprocessing . forking import Popen , duplicate , close , ForkingPickler <EOL> from multiprocessing . util import register_after_fork , debug , sub_debug <EOL> from multiprocessing . connection import Client , Listener <EOL> if not ( sys . platform == '<STR_LIT:win32>' or hasattr ( _multiprocessing , '<STR_LIT>' ) ) : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> import _subprocess <EOL> from . _multiprocessing import win32 <EOL> def send_handle ( conn , handle , destination_pid ) : <EOL> process_handle = win32 . OpenProcess ( <EOL> win32 . PROCESS_ALL_ACCESS , False , destination_pid <EOL> ) <EOL> try : <EOL> new_handle = duplicate ( handle , process_handle ) <EOL> conn . send ( new_handle ) <EOL> finally : <EOL> close ( process_handle ) <EOL> def recv_handle ( conn ) : <EOL> return conn . recv ( ) <EOL> else : <EOL> def send_handle ( conn , handle , destination_pid ) : <EOL> _multiprocessing . sendfd ( conn . fileno ( ) , handle ) <EOL> def recv_handle ( conn ) : <EOL> return _multiprocessing . recvfd ( conn . fileno ( ) ) <EOL> _cache = set ( ) <EOL> def _reset ( obj ) : <EOL> global _lock , _listener , _cache <EOL> for h in _cache : <EOL> close ( h ) <EOL> _cache . clear ( ) <EOL> _lock = threading . Lock ( ) <EOL> _listener = None <EOL> _reset ( None ) <EOL> register_after_fork ( _reset , _reset ) <EOL> def _get_listener ( ) : <EOL> global _listener <EOL> if _listener is None : <EOL> _lock . acquire ( ) <EOL> try : <EOL> if _listener is None : <EOL> debug ( '<STR_LIT>' ) <EOL> _listener = Listener ( authkey = current_process ( ) . authkey ) <EOL> t = threading . Thread ( target = _serve ) <EOL> t . daemon = True <EOL> t . start ( ) <EOL> finally : <EOL> _lock . release ( ) <EOL> return _listener <EOL> def _serve ( ) : <EOL> from . util import is_exiting , sub_warning <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> conn = _listener . accept ( ) <EOL> handle_wanted , destination_pid = conn . recv ( ) <EOL> _cache . remove ( handle_wanted ) <EOL> send_handle ( conn , handle_wanted , destination_pid ) <EOL> close ( handle_wanted ) <EOL> conn . close ( ) <EOL> except : <EOL> if not is_exiting ( ) : <EOL> import traceback <EOL> sub_warning ( <EOL> '<STR_LIT>' + <EOL> '<STR_LIT:->' * <NUM_LIT> + '<STR_LIT:\n>' + traceback . format_exc ( ) + '<STR_LIT:->' * <NUM_LIT> <EOL> ) <EOL> def reduce_handle ( handle ) : <EOL> if Popen . thread_is_spawning ( ) : <EOL> return ( None , Popen . duplicate_for_child ( handle ) , True ) <EOL> dup_handle = duplicate ( handle ) <EOL> _cache . add ( dup_handle ) <EOL> sub_debug ( '<STR_LIT>' , handle ) <EOL> return ( _get_listener ( ) . address , dup_handle , False ) <EOL> def rebuild_handle ( pickled_data ) : <EOL> address , handle , inherited = pickled_data <EOL> if inherited : <EOL> return handle <EOL> sub_debug ( '<STR_LIT>' , handle ) <EOL> conn = Client ( address , authkey = current_process ( ) . authkey ) <EOL> conn . send ( ( handle , os . getpid ( ) ) ) <EOL> new_handle = recv_handle ( conn ) <EOL> conn . close ( ) <EOL> return new_handle <EOL> def reduce_connection ( conn ) : <EOL> rh = reduce_handle ( conn . fileno ( ) ) <EOL> return rebuild_connection , ( rh , conn . readable , conn . writable ) <EOL> def rebuild_connection ( reduced_handle , readable , writable ) : <EOL> handle = rebuild_handle ( reduced_handle ) <EOL> return _multiprocessing . Connection ( <EOL> handle , readable = readable , writable = writable <EOL> ) <EOL> ForkingPickler . register ( _multiprocessing . Connection , reduce_connection ) <EOL> def fromfd ( fd , family , type_ , proto = <NUM_LIT:0> ) : <EOL> s = socket . fromfd ( fd , family , type_ , proto ) <EOL> if s . __class__ is not socket . socket : <EOL> s = socket . socket ( _sock = s ) <EOL> return s <EOL> def reduce_socket ( s ) : <EOL> reduced_handle = reduce_handle ( s . fileno ( ) ) <EOL> return rebuild_socket , ( reduced_handle , s . family , s . type , s . proto ) <EOL> def rebuild_socket ( reduced_handle , family , type_ , proto ) : <EOL> fd = rebuild_handle ( reduced_handle ) <EOL> _sock = fromfd ( fd , family , type_ , proto ) <EOL> close ( fd ) <EOL> return _sock <EOL> ForkingPickler . register ( socket . socket , reduce_socket ) <EOL> if sys . platform == '<STR_LIT:win32>' : <EOL> def reduce_pipe_connection ( conn ) : <EOL> rh = reduce_handle ( conn . fileno ( ) ) <EOL> return rebuild_pipe_connection , ( rh , conn . readable , conn . writable ) <EOL> def rebuild_pipe_connection ( reduced_handle , readable , writable ) : <EOL> handle = rebuild_handle ( reduced_handle ) <EOL> return _multiprocessing . PipeConnection ( <EOL> handle , readable = readable , writable = writable <EOL> ) <EOL> ForkingPickler . register ( _multiprocessing . PipeConnection , reduce_pipe_connection ) </s>
