<s> """<STR_LIT>""" <EOL> import logging <EOL> from thespian . actors import * <EOL> from thespian . system . systemBase import systemBase <EOL> from thespian . system . utilis import thesplog , checkActorCapabilities , partition <EOL> from thespian . system . transport import * <EOL> from thespian . system . logdirector import * <EOL> from thespian . system . utilis import setProcName , StatsManager <EOL> from thespian . system . addressManager import ActorLocalAddress <EOL> from thespian . system . messages . multiproc import * <EOL> from thespian . system . sourceLoader import loadModuleFromHashSource <EOL> from functools import partial <EOL> import multiprocessing <EOL> import signal <EOL> from datetime import timedelta <EOL> MAX_ADMIN_STARTUP_DELAY = timedelta ( seconds = <NUM_LIT:5> ) <EOL> uncatchable_signals = [ ] <EOL> for sname in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> try : <EOL> uncatchable_signals . append ( eval ( '<STR_LIT>' % sname ) ) <EOL> except AttributeError : <EOL> pass <EOL> exit_signals = [ ] <EOL> for sname in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> try : <EOL> exit_signals . append ( eval ( '<STR_LIT>' % sname ) ) <EOL> except AttributeError : <EOL> pass <EOL> child_exit_signals = [ ] <EOL> for sname in [ '<STR_LIT>' ] : <EOL> try : <EOL> child_exit_signals . append ( eval ( '<STR_LIT>' % sname ) ) <EOL> except AttributeError : <EOL> pass <EOL> def detach_child ( childref ) : <EOL> if hasattr ( multiprocessing . process , '<STR_LIT>' ) : <EOL> multiprocessing . process . _children . remove ( childref ) <EOL> if hasattr ( multiprocessing . process , '<STR_LIT>' ) : <EOL> if hasattr ( multiprocessing . process . _current_process , '<STR_LIT>' ) : <EOL> multiprocessing . process . _current_process . _children . remove ( childref ) <EOL> class multiprocessCommon ( systemBase ) : <EOL> def __init__ ( self , system , logDefs = None ) : <EOL> import sys , time <EOL> system . capabilities [ '<STR_LIT>' ] = tuple ( sys . version_info ) <EOL> system . capabilities [ '<STR_LIT>' ] = ThespianGeneration <EOL> system . capabilities [ '<STR_LIT>' ] = str ( int ( time . time ( ) * <NUM_LIT:1000> ) ) <EOL> self . transport = self . transportType ( ExternalInterfaceTransportInit ( ) , <EOL> system . capabilities , logDefs ) <EOL> super ( multiprocessCommon , self ) . __init__ ( system , logDefs ) <EOL> def _startAdmin ( self , adminAddr , addrOfStarter , capabilities , logDefs ) : <EOL> endpointPrep = self . transport . prepEndpoint ( adminAddr , capabilities ) <EOL> multiprocessing . process . _current_process . _daemonic = False <EOL> admin = multiprocessing . Process ( target = startAdmin , <EOL> args = ( MultiProcAdmin , <EOL> addrOfStarter , <EOL> endpointPrep , <EOL> self . transport . __class__ , <EOL> adminAddr , <EOL> capabilities , <EOL> logDefs ) , <EOL> name = '<STR_LIT>' ) <EOL> admin . start ( ) <EOL> detach_child ( admin ) <EOL> self . transport . connectEndpoint ( endpointPrep ) <EOL> response = self . transport . run ( None , MAX_ADMIN_STARTUP_DELAY ) <EOL> if not response or not isinstance ( response . message , EndpointConnected ) : <EOL> raise InvalidActorAddress ( adminAddr , '<STR_LIT>' ) <EOL> def closeUnusedFiles ( transport ) : <EOL> import os , sys <EOL> notouch = transport . protectedFileNumList ( ) <EOL> for each in [ sys . stdin , sys . stderr , sys . stdout ] : <EOL> try : <EOL> notouch . append ( each . fileno ( ) ) <EOL> except AttributeError : pass <EOL> for fdnum in range ( <NUM_LIT:3> , <NUM_LIT:255> ) : <EOL> if fdnum not in notouch : <EOL> try : <EOL> os . close ( fdnum ) <EOL> except OSError : pass <EOL> def closeFileNums ( list ) : <EOL> import os <EOL> for fdnum in list : <EOL> try : <EOL> os . close ( fdnum ) <EOL> except OSError : pass <EOL> from thespian . system . systemAdmin import ThespianAdmin <EOL> def startAdmin ( adminClass , addrOfStarter , endpointPrep , transportClass , <EOL> adminAddr , capabilities , logDefs ) : <EOL> import os , sys <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> os . setsid ( ) <EOL> try : <EOL> import resource <EOL> resource . setrlimit ( resource . RLIMIT_CORE , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> except Exception : pass <EOL> if hasattr ( os , '<STR_LIT>' ) : <EOL> if os . fork ( ) : sys . exit ( <NUM_LIT:0> ) <EOL> transport = transportClass ( endpointPrep ) <EOL> try : <EOL> admin = adminClass ( transport , adminAddr , capabilities , logDefs ) <EOL> except Exception : <EOL> transport . scheduleTransmit ( None , <EOL> TransmitIntent ( addrOfStarter , EndpointConnected ( <NUM_LIT:0> ) ) ) <EOL> raise <EOL> admin . addrOfStarter = addrOfStarter <EOL> setProcName ( adminClass . __name__ , admin . transport . myAddress ) <EOL> loggerAddr = ActorAddress ( ActorLocalAddress ( transport . myAddress , - <NUM_LIT:1> , None ) ) <EOL> admin . asLogger = None <EOL> logAggregator = capabilities . get ( '<STR_LIT>' , None ) <EOL> if logAggregator : <EOL> try : <EOL> logAggregator = transportClass . getAddressFromString ( logAggregator ) <EOL> except Exception as ex : <EOL> thesplog ( '<STR_LIT>' , <EOL> logAggregator , ex , level = logging . WARNING ) <EOL> admin . asLogProc = startASLogger ( loggerAddr , logDefs , transport , capabilities , <EOL> logAggregator <EOL> if logAggregator != admin . transport . myAddress <EOL> else None ) <EOL> admin . run ( ) <EOL> class ChildInfo ( object ) : <EOL> def __init__ ( self , childAddr , childClass , childProc , childNum ) : <EOL> self . childAddr = childAddr <EOL> self . childClass = childClass <EOL> self . childProc = childProc <EOL> self . childNum = childNum <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( str ( self . childNum ) , <EOL> str ( self . childClass ) , <EOL> str ( getattr ( self , '<STR_LIT>' , self . childAddr ) ) , <EOL> str ( self . childProc ) ) <EOL> def startASLogger ( loggerAddr , logDefs , transport , capabilities , aggregatorAddress = None ) : <EOL> endpointPrep = transport . prepEndpoint ( loggerAddr , capabilities ) <EOL> multiprocessing . process . _current_process . _daemonic = False <EOL> logProc = multiprocessing . Process ( target = startupASLogger , <EOL> args = ( transport . myAddress , endpointPrep , <EOL> logDefs , <EOL> transport . __class__ , aggregatorAddress ) ) <EOL> logProc . daemon = True <EOL> logProc . start ( ) <EOL> transport . connectEndpoint ( endpointPrep ) <EOL> return ChildInfo ( loggerAddr , '<STR_LIT>' , logProc , endpointPrep . addrInst ) <EOL> class MultiProcReplicator ( object ) : <EOL> def _startChildActor ( self , childAddr , childClass , parentAddr , notifyAddr , <EOL> childRequirements = None , <EOL> sourceHash = None , sourceToLoad = None ) : <EOL> """<STR_LIT>""" <EOL> if parentAddr is None : <EOL> raise ActorSystemFailure ( '<STR_LIT>' ) <EOL> if self . asLogger is None : <EOL> raise ActorSystemFailure ( '<STR_LIT>' ) <EOL> if not checkActorCapabilities ( childClass , self . capabilities , childRequirements , <EOL> partial ( loadModuleFromHashSource , <EOL> sourceHash , <EOL> { sourceHash : sourceToLoad } ) <EOL> if sourceHash else None ) : <EOL> raise NoCompatibleSystemForActor ( childClass , <EOL> "<STR_LIT>" ) <EOL> endpointPrep = self . transport . prepEndpoint ( childAddr , self . capabilities ) <EOL> multiprocessing . process . _current_process . _daemonic = False <EOL> fileNumsToClose = list ( self . transport . childResetFileNumList ( ) ) <EOL> child = multiprocessing . Process ( target = startChild , <EOL> args = ( childClass , <EOL> endpointPrep , <EOL> self . transport . __class__ , <EOL> sourceHash or self . _sourceHash , <EOL> sourceToLoad , <EOL> parentAddr , <EOL> self . _adminAddr , <EOL> notifyAddr , <EOL> self . asLogger , <EOL> childRequirements , <EOL> self . capabilities , <EOL> fileNumsToClose ) , <EOL> name = '<STR_LIT>' % ( getattr ( childClass , '<STR_LIT>' , childClass ) , str ( childAddr ) ) ) <EOL> child . start ( ) <EOL> detach_child ( child ) <EOL> if not hasattr ( self , '<STR_LIT>' ) : self . _child_procs = [ ] <EOL> self . _child_procs . append ( ChildInfo ( childAddr , childClass , child , endpointPrep . addrInst ) ) <EOL> self . transport . connectEndpoint ( endpointPrep ) <EOL> @ staticmethod <EOL> def _checkChildLiveness ( childInfo ) : <EOL> if not childInfo . childProc . is_alive ( ) : <EOL> childInfo . childProc . join ( <NUM_LIT:0.5> ) <EOL> return False <EOL> return True <EOL> def _childExited ( self , childAddr ) : <EOL> self . _child_procs = list ( filter ( self . _checkChildLiveness , getattr ( self , '<STR_LIT>' , [ ] ) ) ) <EOL> def childDied ( self , signum , frame ) : <EOL> self . _child_procs , dead = partition ( self . _checkChildLiveness , getattr ( self , '<STR_LIT>' , [ ] ) ) <EOL> for each in dead : <EOL> addr = getattr ( each , '<STR_LIT>' , each . childAddr ) <EOL> self . transport . scheduleTransmit ( None , TransmitIntent ( self . transport . myAddress , <EOL> ChildActorExited ( addr ) ) ) <EOL> def h_EndpointConnected ( self , envelope ) : <EOL> for C in getattr ( self , '<STR_LIT>' , [ ] ) : <EOL> if envelope . message . childInstance == C . childNum : <EOL> C . childRealAddr = envelope . sender <EOL> break <EOL> else : <EOL> thesplog ( '<STR_LIT>' , envelope , level = logging . WARNING ) <EOL> self . _pendingActorReady ( envelope . message . childInstance , envelope . sender ) <EOL> return True <EOL> def h_LoggerConnected ( self , envelope ) : <EOL> self . asLogger = envelope . sender <EOL> self . oldLoggerRoot = logging . root <EOL> logging . root = ThespianLogForwarder ( self . asLogger , self . transport ) <EOL> logging . Logger . root = logging . root <EOL> logging . Logger . manager = logging . Manager ( logging . Logger . root ) <EOL> logging . getLogger ( '<STR_LIT>' ) . info ( '<STR_LIT>' , self . myAddress ) <EOL> self . transport . scheduleTransmit ( None , <EOL> TransmitIntent ( self . addrOfStarter , EndpointConnected ( <NUM_LIT:0> ) ) ) <EOL> self . _activate ( ) <EOL> return True <EOL> def h_LogRecord ( self , envelope ) : <EOL> self . _send_intent ( TransmitIntent ( self . asLogger , envelope . message ) ) <EOL> return True <EOL> def _handleReplicatorMessages ( self , envelope ) : <EOL> if isinstance ( envelope . message , EndpointConnected ) : <EOL> return True , self . h_EndpointConnected ( envelope ) <EOL> if isinstance ( envelope . message , logging . LogRecord ) : <EOL> return True , self . h_LogRecord ( envelope ) <EOL> return False , True <EOL> def _cleanupAdmin ( self ) : <EOL> if getattr ( self , '<STR_LIT>' , None ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> logging . root = self . oldLoggerRoot <EOL> logging . Logger . root = self . oldLoggerRoot <EOL> logging . Logger . manager = logging . Manager ( logging . Logger . root ) <EOL> self . transport . run ( TransmitOnly , maximumDuration = timedelta ( milliseconds = <NUM_LIT> ) ) <EOL> import time <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . transport . scheduleTransmit ( None , TransmitIntent ( self . asLogger , <EOL> LoggerExitRequest ( ) ) ) <EOL> self . transport . run ( TransmitOnly ) <EOL> if getattr ( self , '<STR_LIT>' , None ) : <EOL> if self . _checkChildLiveness ( self . asLogProc ) : <EOL> import time <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . _checkChildLiveness ( self . asLogProc ) <EOL> from thespian . system . actorManager import ActorManager <EOL> def signal_detector ( name , addr ) : <EOL> def signal_detected ( signum , frame ) : <EOL> thesplog ( '<STR_LIT>' , name , addr , signum , <EOL> level = logging . WARNING ) <EOL> return signal_detected <EOL> def shutdown_signal_detector ( name , addr , am ) : <EOL> def shutdown_signal_detected ( signum , frame ) : <EOL> thesplog ( '<STR_LIT>' , name , addr , signum , <EOL> level = logging . WARNING ) <EOL> am . transport . scheduleTransmit ( None , TransmitIntent ( am . transport . myAddress , <EOL> ActorExitRequest ( ) ) ) <EOL> return shutdown_signal_detected <EOL> def startChild ( childClass , endpoint , transportClass , <EOL> sourceHash , sourceToLoad , <EOL> parentAddr , adminAddr , notifyAddr , loggerAddr , <EOL> childRequirements , currentSystemCapabilities , <EOL> fileNumsToClose ) : <EOL> closeFileNums ( fileNumsToClose ) <EOL> multiprocessing . process . _current_process . _daemonic = False <EOL> transport = transportClass ( endpoint ) <EOL> logging . root = ThespianLogForwarder ( loggerAddr , transport ) <EOL> logging . Logger . root = logging . root <EOL> logging . Logger . manager = logging . Manager ( logging . Logger . root ) <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> am = MultiProcManager ( childClass , transport , <EOL> sourceHash , sourceToLoad , <EOL> parentAddr , adminAddr , <EOL> childRequirements , currentSystemCapabilities ) <EOL> am . asLogger = loggerAddr <EOL> am . transport . scheduleTransmit ( None , <EOL> TransmitIntent ( notifyAddr , <EOL> EndpointConnected ( endpoint . addrInst ) ) ) <EOL> setProcName ( getattr ( childClass , '<STR_LIT>' , str ( childClass ) ) , am . transport . myAddress ) <EOL> sighandler = signal_detector ( getattr ( childClass , '<STR_LIT>' , str ( childClass ) ) , <EOL> am . transport . myAddress ) <EOL> sigexithandler = shutdown_signal_detector ( getattr ( childClass , '<STR_LIT>' , str ( childClass ) ) , <EOL> am . transport . myAddress , <EOL> am ) <EOL> for each in range ( <NUM_LIT:1> , signal . NSIG ) : <EOL> if each not in uncatchable_signals : <EOL> if each in child_exit_signals : <EOL> signal . signal ( each , am . childDied ) <EOL> continue <EOL> try : <EOL> signal . signal ( each , <EOL> sigexithandler if each in exit_signals else sighandler ) <EOL> except ( RuntimeError , ValueError , EnvironmentError ) as ex : <EOL> pass <EOL> am . run ( ) <EOL> class MultiProcAdmin ( MultiProcReplicator , ThespianAdmin ) : pass <EOL> class MultiProcManager ( MultiProcReplicator , ActorManager ) : pass </s>
