<s> """<STR_LIT>""" <EOL> import time <EOL> import threading <EOL> import lxml . etree <EOL> import multiprocessing <EOL> import Bcfg2 . Options <EOL> import Bcfg2 . Server . Cache <EOL> import Bcfg2 . Server . Plugin <EOL> from itertools import cycle <EOL> from Bcfg2 . Compat import Queue , Empty , wraps <EOL> from Bcfg2 . Server . Core import Core , exposed <EOL> from Bcfg2 . Server . BuiltinCore import BuiltinCore <EOL> from multiprocessing . connection import Listener , Client <EOL> class RPCQueue ( Bcfg2 . Server . Plugin . Debuggable ) : <EOL> """<STR_LIT>""" <EOL> poll_wait = <NUM_LIT> <EOL> def __init__ ( self ) : <EOL> Bcfg2 . Server . Plugin . Debuggable . __init__ ( self ) <EOL> self . _terminate = threading . Event ( ) <EOL> self . _queues = dict ( ) <EOL> self . _listeners = [ ] <EOL> def add_subscriber ( self , name ) : <EOL> """<STR_LIT>""" <EOL> self . _queues [ name ] = multiprocessing . Queue ( ) <EOL> return self . _queues [ name ] <EOL> def publish ( self , method , args = None , kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> for queue in self . _queues . values ( ) : <EOL> queue . put ( ( None , ( method , args or [ ] , kwargs or dict ( ) ) ) ) <EOL> def rpc ( self , dest , method , args = None , kwargs = None ) : <EOL> """<STR_LIT>""" <EOL> listener = Listener ( ) <EOL> self . logger . debug ( "<STR_LIT>" % listener . address ) <EOL> self . _listeners . append ( listener ) <EOL> try : <EOL> self . _queues [ dest ] . put ( ( listener . address , <EOL> ( method , args or [ ] , kwargs or dict ( ) ) ) ) <EOL> conn = listener . accept ( ) <EOL> try : <EOL> while not self . _terminate . is_set ( ) : <EOL> if conn . poll ( self . poll_wait ) : <EOL> return conn . recv ( ) <EOL> finally : <EOL> conn . close ( ) <EOL> finally : <EOL> listener . close ( ) <EOL> self . _listeners . remove ( listener ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _terminate . set ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> for name , queue in self . _queues . items ( ) : <EOL> self . logger . debug ( "<STR_LIT>" % name ) <EOL> queue . close ( ) <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> for listener in self . _listeners : <EOL> self . logger . debug ( "<STR_LIT>" % <EOL> listener . address ) <EOL> listener . close ( ) <EOL> class DualEvent ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , threading_event = None , multiprocessing_event = None ) : <EOL> self . _threading_event = threading_event or threading . Event ( ) <EOL> self . _multiproc_event = multiprocessing_event or multiprocessing . Event ( ) <EOL> if threading_event or multiprocessing_event : <EOL> self . clear ( ) <EOL> def is_set ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _threading_event . is_set ( ) <EOL> isSet = is_set <EOL> def set ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _threading_event . set ( ) <EOL> self . _multiproc_event . set ( ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _threading_event . clear ( ) <EOL> self . _multiproc_event . clear ( ) <EOL> def wait ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _threading_event . wait ( timeout = timeout ) <EOL> class ChildCore ( Core ) : <EOL> """<STR_LIT>""" <EOL> poll_wait = <NUM_LIT> <EOL> def __init__ ( self , name , rpc_q , terminate ) : <EOL> """<STR_LIT>""" <EOL> Core . __init__ ( self ) <EOL> self . name = name <EOL> self . terminate = terminate <EOL> self . rpc_q = rpc_q <EOL> Bcfg2 . Options . setup . daemon = False <EOL> self . perflog_thread = None <EOL> self . _rmi = dict ( ) <EOL> def _run ( self ) : <EOL> return True <EOL> def _dispatch ( self , address , data ) : <EOL> """<STR_LIT>""" <EOL> if address is not None : <EOL> self . logger . debug ( "<STR_LIT>" % address ) <EOL> client = Client ( address ) <EOL> method , args , kwargs = data <EOL> func = None <EOL> rv = None <EOL> if "<STR_LIT:.>" in method : <EOL> if method in self . _rmi : <EOL> func = self . _rmi [ method ] <EOL> else : <EOL> self . logger . error ( "<STR_LIT>" % ( self . name , <EOL> method ) ) <EOL> elif not hasattr ( self , method ) : <EOL> self . logger . error ( "<STR_LIT>" % ( self . name , <EOL> method ) ) <EOL> else : <EOL> func = getattr ( self , method ) <EOL> if not func . exposed : <EOL> self . logger . error ( "<STR_LIT>" % ( self . name , <EOL> method ) ) <EOL> func = None <EOL> if func is not None : <EOL> self . logger . debug ( "<STR_LIT>" % ( self . name , <EOL> method ) ) <EOL> rv = func ( * args , ** kwargs ) <EOL> if address is not None : <EOL> self . logger . debug ( "<STR_LIT>" % address ) <EOL> client . send ( rv ) <EOL> def _block ( self ) : <EOL> self . _rmi = self . _get_rmi ( ) <EOL> while not self . terminate . is_set ( ) : <EOL> try : <EOL> address , data = self . rpc_q . get ( timeout = self . poll_wait ) <EOL> threadname = "<STR_LIT:->" . join ( str ( i ) for i in data ) <EOL> rpc_thread = threading . Thread ( name = threadname , <EOL> target = self . _dispatch , <EOL> args = [ address , data ] ) <EOL> rpc_thread . start ( ) <EOL> except Empty : <EOL> pass <EOL> except KeyboardInterrupt : <EOL> break <EOL> self . shutdown ( ) <EOL> def shutdown ( self ) : <EOL> Core . shutdown ( self ) <EOL> self . logger . info ( "<STR_LIT>" % self . name ) <EOL> self . rpc_q . close ( ) <EOL> while len ( threading . enumerate ( ) ) > <NUM_LIT:1> : <EOL> threads = [ t for t in threading . enumerate ( ) <EOL> if t != threading . current_thread ( ) ] <EOL> self . logger . info ( "<STR_LIT>" % <EOL> ( self . name , len ( threads ) , <EOL> [ t . name for t in threads ] ) ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . logger . info ( "<STR_LIT>" % self . name ) <EOL> def _get_rmi ( self ) : <EOL> rmi = dict ( ) <EOL> for pname , pinst in self . _get_rmi_objects ( ) . items ( ) : <EOL> for crmi in pinst . __child_rmi__ : <EOL> if isinstance ( crmi , tuple ) : <EOL> mname = crmi [ <NUM_LIT:1> ] <EOL> else : <EOL> mname = crmi <EOL> rmi [ "<STR_LIT>" % ( pname , mname ) ] = getattr ( pinst , mname ) <EOL> return rmi <EOL> @ exposed <EOL> def expire_cache ( self , * tags , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> Bcfg2 . Server . Cache . expire ( * tags , exact = kwargs . pop ( "<STR_LIT>" , False ) ) <EOL> @ exposed <EOL> def GetConfig ( self , client ) : <EOL> """<STR_LIT>""" <EOL> self . metadata . update_client_list ( ) <EOL> self . logger . debug ( "<STR_LIT>" % <EOL> ( self . name , client ) ) <EOL> return lxml . etree . tostring ( self . BuildConfiguration ( client ) ) <EOL> class MultiprocessingCore ( BuiltinCore ) : <EOL> """<STR_LIT>""" <EOL> options = BuiltinCore . options + [ <EOL> Bcfg2 . Options . Option ( <EOL> '<STR_LIT>' , dest = "<STR_LIT>" , <EOL> cf = ( '<STR_LIT>' , '<STR_LIT>' ) , type = int , <EOL> default = multiprocessing . cpu_count ( ) , <EOL> help = '<STR_LIT>' ) ] <EOL> shutdown_timeout = <NUM_LIT> <EOL> def __init__ ( self ) : <EOL> BuiltinCore . __init__ ( self ) <EOL> self . pipes = dict ( ) <EOL> self . available_children = Queue ( maxsize = Bcfg2 . Options . setup . core_children ) <EOL> self . terminate = DualEvent ( threading_event = self . terminate ) <EOL> self . rpc_q = RPCQueue ( ) <EOL> self . _all_children = [ ] <EOL> self . children = None <EOL> def __str__ ( self ) : <EOL> if hasattr ( Bcfg2 . Options . setup , "<STR_LIT>" ) : <EOL> return "<STR_LIT>" % ( self . __class__ . __name__ , <EOL> Bcfg2 . Options . setup . server , <EOL> len ( self . _all_children ) ) <EOL> else : <EOL> return "<STR_LIT>" % ( self . __class__ . __name__ , <EOL> len ( self . _all_children ) ) <EOL> def _run ( self ) : <EOL> for cnum in range ( Bcfg2 . Options . setup . core_children ) : <EOL> name = "<STR_LIT>" % cnum <EOL> self . logger . debug ( "<STR_LIT>" % name ) <EOL> child_q = self . rpc_q . add_subscriber ( name ) <EOL> childcore = ChildCore ( name , child_q , self . terminate ) <EOL> child = multiprocessing . Process ( target = childcore . run , name = name ) <EOL> child . start ( ) <EOL> self . logger . debug ( "<STR_LIT>" % ( name , <EOL> child . pid ) ) <EOL> self . _all_children . append ( name ) <EOL> self . logger . debug ( "<STR_LIT>" % ( len ( self . _all_children ) , <EOL> self . _all_children ) ) <EOL> self . children = cycle ( self . _all_children ) <EOL> Bcfg2 . Server . Cache . add_expire_hook ( self . cache_dispatch ) <EOL> return BuiltinCore . _run ( self ) <EOL> def shutdown ( self ) : <EOL> BuiltinCore . shutdown ( self ) <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> self . rpc_q . close ( ) <EOL> def term_children ( ) : <EOL> """<STR_LIT>""" <EOL> for child in multiprocessing . active_children ( ) : <EOL> self . logger . error ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . shutdown_timeout , <EOL> child . name ) ) <EOL> child . terminate ( ) <EOL> timer = threading . Timer ( self . shutdown_timeout , term_children ) <EOL> timer . start ( ) <EOL> while len ( multiprocessing . active_children ( ) ) : <EOL> self . logger . info ( "<STR_LIT>" % <EOL> ( len ( multiprocessing . active_children ( ) ) , <EOL> [ c . name <EOL> for c in multiprocessing . active_children ( ) ] ) ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> timer . cancel ( ) <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> while len ( threading . enumerate ( ) ) > <NUM_LIT:1> : <EOL> threads = [ t for t in threading . enumerate ( ) <EOL> if t != threading . current_thread ( ) ] <EOL> self . logger . info ( "<STR_LIT>" % <EOL> ( len ( threads ) , [ t . name for t in threads ] ) ) <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> def _get_rmi ( self ) : <EOL> child_rmi = dict ( ) <EOL> for pname , pinst in self . _get_rmi_objects ( ) . items ( ) : <EOL> for crmi in pinst . __child_rmi__ : <EOL> if isinstance ( crmi , tuple ) : <EOL> parentname , childname = crmi <EOL> else : <EOL> parentname = childname = crmi <EOL> child_rmi [ "<STR_LIT>" % ( pname , parentname ) ] = "<STR_LIT>" % ( pname , childname ) <EOL> rmi = BuiltinCore . _get_rmi ( self ) <EOL> for method in rmi . keys ( ) : <EOL> if method in child_rmi : <EOL> rmi [ method ] = self . _child_rmi_wrapper ( method , <EOL> rmi [ method ] , <EOL> child_rmi [ method ] ) <EOL> return rmi <EOL> def _child_rmi_wrapper ( self , method , parent_rmi , child_rmi ) : <EOL> """<STR_LIT>""" <EOL> @ wraps ( parent_rmi ) <EOL> def inner ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . logger . debug ( "<STR_LIT>" % <EOL> ( method , child_rmi ) ) <EOL> self . rpc_q . publish ( child_rmi , args = args , kwargs = kwargs ) <EOL> return parent_rmi ( * args , ** kwargs ) <EOL> return inner <EOL> @ exposed <EOL> def set_debug ( self , address , debug ) : <EOL> self . rpc_q . set_debug ( debug ) <EOL> self . rpc_q . publish ( "<STR_LIT>" , args = [ address , debug ] ) <EOL> return BuiltinCore . set_debug ( self , address , debug ) <EOL> def cache_dispatch ( self , tags , exact , _ ) : <EOL> """<STR_LIT>""" <EOL> self . rpc_q . publish ( "<STR_LIT>" , args = tags , kwargs = dict ( exact = exact ) ) <EOL> @ exposed <EOL> def GetConfig ( self , address ) : <EOL> client = self . resolve_client ( address ) [ <NUM_LIT:0> ] <EOL> childname = self . children . next ( ) <EOL> self . logger . debug ( "<STR_LIT>" % ( client , <EOL> childname ) ) <EOL> return self . rpc_q . rpc ( childname , "<STR_LIT>" , args = [ client ] ) <EOL> @ exposed <EOL> def get_statistics ( self , address ) : <EOL> stats = dict ( ) <EOL> def _aggregate_statistics ( newstats , prefix = None ) : <EOL> """<STR_LIT>""" <EOL> for statname , vals in newstats . items ( ) : <EOL> if statname . startswith ( "<STR_LIT>" ) : <EOL> statname = statname [ <NUM_LIT:5> : ] <EOL> if prefix : <EOL> prettyname = "<STR_LIT>" % ( prefix , statname ) <EOL> else : <EOL> prettyname = statname <EOL> stats [ prettyname ] = vals <EOL> totalname = "<STR_LIT>" % statname <EOL> if totalname not in stats : <EOL> stats [ totalname ] = vals <EOL> else : <EOL> newmin = min ( stats [ totalname ] [ <NUM_LIT:0> ] , vals [ <NUM_LIT:0> ] ) <EOL> newmax = max ( stats [ totalname ] [ <NUM_LIT:1> ] , vals [ <NUM_LIT:1> ] ) <EOL> newcount = stats [ totalname ] [ <NUM_LIT:3> ] + vals [ <NUM_LIT:3> ] <EOL> newmean = ( ( stats [ totalname ] [ <NUM_LIT:2> ] * stats [ totalname ] [ <NUM_LIT:3> ] ) + <EOL> ( vals [ <NUM_LIT:2> ] * vals [ <NUM_LIT:3> ] ) ) / newcount <EOL> stats [ totalname ] = ( newmin , newmax , newmean , newcount ) <EOL> stats = dict ( ) <EOL> for childname in self . _all_children : <EOL> _aggregate_statistics ( <EOL> self . rpc_q . rpc ( childname , "<STR_LIT>" , args = [ address ] ) , <EOL> prefix = childname ) <EOL> _aggregate_statistics ( BuiltinCore . get_statistics ( self , address ) ) <EOL> return stats </s>
