<s> "<STR_LIT>" <EOL> import logging <EOL> from thespian . actors import * <EOL> from thespian . system . utilis import thesplog , StatsManager <EOL> from thespian . system . ratelimit import RateThrottle <EOL> from thespian . system . addressManager import ActorAddressManager , CannotPickleAddress <EOL> from thespian . system . messages . logcontrol import SetLogging <EOL> from thespian . system . transport import * <EOL> RATE_THROTTLE = ( lambda sizeAvg , linkSpeed , percentage : <EOL> int ( ( linkSpeed / sizeAvg ) * ( percentage / <NUM_LIT> ) ) ) ( <NUM_LIT> , <NUM_LIT> * <NUM_LIT> * <NUM_LIT> , <NUM_LIT> ) <EOL> class systemCommonBase ( object ) : <EOL> def __init__ ( self , adminAddr , transport ) : <EOL> self . _adminAddr = adminAddr <EOL> self . transport = transport <EOL> self . _addrManager = ActorAddressManager ( adminAddr , self . transport . myAddress ) <EOL> self . transport . setAddressManager ( self . _addrManager ) <EOL> self . _finalTransmitPending = { } <EOL> self . _awaitingAddressUpdate = { } <EOL> self . _receiveQueue = [ ] <EOL> self . _children = [ ] <EOL> self . _governer = RateThrottle ( RATE_THROTTLE ) <EOL> self . _sCBStats = StatsManager ( ) <EOL> @ property <EOL> def address ( self ) : return self . transport . myAddress <EOL> @ property <EOL> def myAddress ( self ) : return self . transport . myAddress <EOL> @ property <EOL> def childAddresses ( self ) : return self . _children <EOL> def _registerChild ( self , childAddress ) : self . _children . append ( childAddress ) <EOL> def _handleChildExited ( self , childAddress ) : <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> self . _addrManager . deadAddress ( childAddress ) <EOL> self . transport . deadAddress ( self . _addrManager , childAddress ) <EOL> self . _childExited ( childAddress ) <EOL> self . _children = [ C for C in self . _children if C != childAddress ] <EOL> if hasattr ( self , '<STR_LIT>' ) and not self . _children : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _exitedAlready = True <EOL> self . _sayGoodbye ( ) <EOL> self . transport . abort_run ( drain = True ) <EOL> return True <EOL> def _updateStatusResponse ( self , resp ) : <EOL> "<STR_LIT>" <EOL> for each in self . childAddresses : <EOL> resp . addChild ( each ) <EOL> for each in self . _receiveQueue : <EOL> resp . addReceivedMessage ( each . sender , self . myAddress , each . message ) <EOL> self . _sCBStats . copyToStatusResponse ( resp ) <EOL> resp . governer = str ( self . _governer ) <EOL> for addr in self . _awaitingAddressUpdate : <EOL> resp . addTXPendingAddressCount ( addr , len ( self . _awaitingAddressUpdate [ addr ] ) ) <EOL> self . transport . _updateStatusResponse ( resp ) <EOL> def setLoggingControls ( self , envelope ) : <EOL> from thespian . system . utilis import thesplog_control <EOL> msg = envelope . message <EOL> thesplog_control ( msg . threshold , msg . useLogging , msg . useFile ) <EOL> return True <EOL> def _send_intent ( self , intent ) : <EOL> self . _governer . eventRatePause ( ) <EOL> sendAddr = self . _addrManager . sendToAddress ( intent . targetAddr ) <EOL> finalIntent = self . _finalTransmitPending . get ( <EOL> intent . targetAddr , <EOL> self . _finalTransmitPending . get ( sendAddr , None ) ) <EOL> self . _finalTransmitPending [ sendAddr or intent . targetAddr ] = intent <EOL> if finalIntent : <EOL> finalIntent . nextIntent = intent <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> return <EOL> self . _send_intent_to_transport ( intent ) <EOL> def _retryPendingChildOperations ( self , childInstance , actualAddress ) : <EOL> lcladdr = self . _addrManager . getLocalAddress ( childInstance ) <EOL> if not actualAddress : <EOL> self . _receiveQueue . append ( ReceiveEnvelope ( lcladdr , ChildActorExited ( lcladdr ) ) ) <EOL> if lcladdr in self . _finalTransmitPending : <EOL> self . _finalTransmitPending [ actualAddress ] = self . _finalTransmitPending [ lcladdr ] <EOL> del self . _finalTransmitPending [ lcladdr ] <EOL> if lcladdr in self . _awaitingAddressUpdate : <EOL> pending = self . _awaitingAddressUpdate [ lcladdr ] <EOL> del self . _awaitingAddressUpdate [ lcladdr ] <EOL> for each in pending : <EOL> if actualAddress : <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> self . _send_intent ( each ) <EOL> else : <EOL> if not isinstance ( each . message , PoisonMessage ) : <EOL> self . _receiveQueue . append ( <EOL> ReceiveEnvelope ( self . myAddress , <EOL> PoisonMessage ( each . message ) ) ) <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> each . result = SendStatus . Failed <EOL> each . completionCallback ( ) <EOL> def _send_intent_to_transport ( self , intent ) : <EOL> thesplog ( '<STR_LIT>' , intent . identify ( ) , level = logging . DEBUG ) <EOL> if not hasattr ( intent , '<STR_LIT>' ) : <EOL> intent . addCallback ( self . _checkNextTransmit , self . _checkNextTransmit ) <EOL> intent . _addedCheckNextTransmitCB = True <EOL> try : <EOL> self . transport . scheduleTransmit ( self . _addrManager , intent ) <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> except CannotPickleAddress as ex : <EOL> thesplog ( '<STR_LIT>' , <EOL> ex . address , hash ( ex . address ) , level = logging . DEBUG ) <EOL> self . _awaitingAddressUpdate . setdefault ( ex . address , [ ] ) . append ( intent ) <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> self . _checkNextTransmit ( <NUM_LIT:0> , intent ) <EOL> except Exception : <EOL> import traceback <EOL> thesplog ( '<STR_LIT>' , intent . identify ( ) , <EOL> traceback . format_exc ( ) , exc_info = True , level = logging . ERROR ) <EOL> if not isinstance ( intent . message , PoisonMessage ) : <EOL> self . _receiveQueue . append ( ReceiveEnvelope ( intent . targetAddr , PoisonMessage ( intent . message ) ) ) <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> intent . result = SendStatus . Failed <EOL> intent . completionCallback ( ) <EOL> def _checkNextTransmit ( self , result , completedIntent ) : <EOL> if completedIntent . nextIntent : <EOL> self . _send_intent_to_transport ( completedIntent . nextIntent ) <EOL> else : <EOL> fkey = completedIntent . targetAddr <EOL> if fkey not in self . _finalTransmitPending : <EOL> fkey = self . _addrManager . sendToAddress ( completedIntent . targetAddr ) <EOL> if fkey not in self . _finalTransmitPending : <EOL> if isinstance ( completedIntent . message , DeadEnvelope ) : <EOL> fkey = completedIntent . message . deadAddress <EOL> if fkey not in self . _finalTransmitPending : <EOL> fkey = self . _addrManager . sendToAddress ( fkey ) <EOL> if fkey in self . _finalTransmitPending : <EOL> if self . _finalTransmitPending [ fkey ] != completedIntent : <EOL> thesplog ( '<STR_LIT>' , <EOL> completedIntent . identify ( ) , <EOL> self . _finalTransmitPending [ fkey ] . identify ( ) , <EOL> level = logging . WARNING ) <EOL> del self . _finalTransmitPending [ fkey ] <EOL> else : <EOL> thesplog ( '<STR_LIT>' , <EOL> completedIntent . identify ( ) , <EOL> str ( self . _addrManager . sendToAddress ( completedIntent . targetAddr ) ) , <EOL> fkey , <EOL> str ( list ( map ( str , self . _finalTransmitPending . keys ( ) ) ) ) , <EOL> level = logging . WARNING ) <EOL> self . _sCBStats . inc ( '<STR_LIT>' ) <EOL> return </s>
