<s> from calvin . actorstore . store import ActorStore <EOL> from calvin . utilities import dynops <EOL> from calvin . runtime . south . plugins . async import async <EOL> from calvin . utilities . calvinlogger import get_logger <EOL> from calvin . utilities . calvin_callback import CalvinCB <EOL> import calvin . requests . calvinresponse as response <EOL> from calvin . utilities . security import Security <EOL> from calvin . actor . actor import ShadowActor <EOL> _log = get_logger ( __name__ ) <EOL> def log_callback ( reply , ** kwargs ) : <EOL> if reply : <EOL> _log . info ( "<STR_LIT>" % ( kwargs [ '<STR_LIT>' ] , reply ) ) <EOL> class ActorManager ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , node ) : <EOL> super ( ActorManager , self ) . __init__ ( ) <EOL> self . actors = { } <EOL> self . node = node <EOL> def _actor_not_found ( self , actor_id ) : <EOL> _log . exception ( "<STR_LIT>" . format ( actor_id ) ) <EOL> raise Exception ( "<STR_LIT>" . format ( actor_id ) ) <EOL> def new ( self , actor_type , args , state = None , prev_connections = None , connection_list = None , callback = None , <EOL> signature = None , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> _log . debug ( "<STR_LIT>" % ( actor_type , args , state , signature ) ) <EOL> _log . analyze ( self . node . id , "<STR_LIT:+>" , { '<STR_LIT>' : actor_type , '<STR_LIT:state>' : state } ) <EOL> try : <EOL> if state : <EOL> a = self . _new_from_state ( actor_type , state ) <EOL> else : <EOL> a = self . _new ( actor_type , args , credentials ) <EOL> except Exception as e : <EOL> _log . exception ( "<STR_LIT>" ) <EOL> raise ( e ) <EOL> a . signature_set ( signature ) <EOL> self . actors [ a . id ] = a <EOL> self . node . storage . add_actor ( a , self . node . id ) <EOL> if prev_connections : <EOL> connection_list = self . _prev_connections_to_connection_list ( prev_connections ) <EOL> self . node . control . log_actor_new ( a . id , a . name , actor_type , isinstance ( a , ShadowActor ) ) <EOL> if connection_list : <EOL> self . connect ( a . id , connection_list , callback = callback ) <EOL> else : <EOL> if callback : <EOL> callback ( status = response . CalvinResponse ( True ) , actor_id = a . id ) <EOL> else : <EOL> return a . id <EOL> def _new_actor ( self , actor_type , actor_id = None , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> if credentials is not None : <EOL> sec = Security ( ) <EOL> sec . set_principal ( credentials ) <EOL> sec . authenticate_principal ( ) <EOL> else : <EOL> sec = None <EOL> ( found , is_primitive , class_ ) = ActorStore ( security = sec ) . lookup ( actor_type ) <EOL> if not found : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT:class>' : class_ } ) <EOL> found = True <EOL> is_primitive = True <EOL> class_ = ShadowActor <EOL> if not found or not is_primitive : <EOL> _log . error ( "<STR_LIT>" % ( actor_type ) ) <EOL> raise Exception ( "<STR_LIT>" ) <EOL> try : <EOL> a = class_ ( actor_type , actor_id = actor_id ) <EOL> except Exception as e : <EOL> _log . exception ( "<STR_LIT>" ) <EOL> _log . error ( "<STR_LIT>" % ( actor_type , class_ . __init__ ) ) <EOL> raise ( e ) <EOL> try : <EOL> a . set_credentials ( credentials , security = sec ) <EOL> a . _calvinsys = self . node . calvinsys ( ) <EOL> a . check_requirements ( ) <EOL> except Exception as e : <EOL> _log . exception ( "<STR_LIT>" ) <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT:class>' : class_ } ) <EOL> a = ShadowActor ( actor_type , actor_id = actor_id ) <EOL> a . set_credentials ( credentials , security = sec ) <EOL> a . _calvinsys = self . node . calvinsys ( ) <EOL> return a <EOL> def _new ( self , actor_type , args , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> a = self . _new_actor ( actor_type , credentials = credentials ) <EOL> human_readable_name = args . pop ( '<STR_LIT:name>' , '<STR_LIT>' ) <EOL> a . name = human_readable_name <EOL> self . node . pm . add_ports_of_actor ( a ) <EOL> a . init ( ** args ) <EOL> a . setup_complete ( ) <EOL> except Exception as e : <EOL> _log . exception ( "<STR_LIT>" ) <EOL> raise ( e ) <EOL> return a <EOL> def _new_from_state ( self , actor_type , state ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> _log . analyze ( self . node . id , "<STR_LIT:+>" , state ) <EOL> credentials = state . pop ( '<STR_LIT>' , None ) <EOL> try : <EOL> state [ '<STR_LIT>' ] . remove ( '<STR_LIT>' ) <EOL> except : <EOL> pass <EOL> a = self . _new_actor ( actor_type , actor_id = state [ '<STR_LIT:id>' ] , credentials = credentials ) <EOL> if '<STR_LIT>' in state : <EOL> args = state . pop ( '<STR_LIT>' ) <EOL> state [ '<STR_LIT>' ] . remove ( '<STR_LIT>' ) <EOL> a . init ( ** args ) <EOL> did_migrate = isinstance ( a , ShadowActor ) <EOL> else : <EOL> did_migrate = True <EOL> a . _set_state ( state ) <EOL> self . node . pm . add_ports_of_actor ( a ) <EOL> if did_migrate : <EOL> a . did_migrate ( ) <EOL> a . setup_complete ( ) <EOL> except Exception as e : <EOL> _log . exception ( "<STR_LIT>" % ( a , dir ( a ) ) ) <EOL> raise ( e ) <EOL> return a <EOL> def destroy ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> self . node . metering . remove_actor_info ( actor_id ) <EOL> a = self . actors [ actor_id ] <EOL> a . will_end ( ) <EOL> self . node . pm . remove_ports_of_actor ( a ) <EOL> self . node . storage . delete_actor ( actor_id ) <EOL> del self . actors [ actor_id ] <EOL> self . node . control . log_actor_destroy ( a . id ) <EOL> def enable ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> self . actors [ actor_id ] . enable ( ) <EOL> def disable ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> _log . info ( "<STR_LIT>" , actor_id ) <EOL> self . _actor_not_found ( actor_id ) <EOL> self . actors [ actor_id ] . disable ( ) <EOL> def update_requirements ( self , actor_id , requirements , extend = False , move = False , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if actor_id not in self . actors : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT>' : actor_id } ) <EOL> if callback : <EOL> callback ( status = response . CalvinResponse ( False ) ) <EOL> return <EOL> if not isinstance ( requirements , ( list , tuple ) ) : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT>' : actor_id } ) <EOL> if callback : <EOL> callback ( status = response . CalvinResponse ( response . BAD_REQUEST ) ) <EOL> return <EOL> actor = self . actors [ actor_id ] <EOL> actor . _collect_placement_counter = <NUM_LIT:0> <EOL> actor . _collect_placement_last_value = <NUM_LIT:0> <EOL> actor . _collect_placement_cb = None <EOL> actor . requirements_add ( requirements , extend ) <EOL> node_iter = self . node . app_manager . actor_requirements ( None , actor_id ) <EOL> possible_placements = set ( [ ] ) <EOL> done = [ False ] <EOL> node_iter . set_cb ( self . _update_requirements_placements , node_iter , actor_id , possible_placements , <EOL> move = move , cb = callback , done = done ) <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT>' : actor_id , '<STR_LIT>' : str ( node_iter ) } ) <EOL> self . _update_requirements_placements ( node_iter , actor_id , possible_placements , <EOL> move = move , cb = callback , done = done ) <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { '<STR_LIT>' : actor_id , '<STR_LIT>' : str ( node_iter ) } ) <EOL> def _update_requirements_placements ( self , node_iter , actor_id , possible_placements , done , move = False , cb = None ) : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { } , tb = True ) <EOL> actor = self . actors [ actor_id ] <EOL> if actor . _collect_placement_cb : <EOL> actor . _collect_placement_cb . cancel ( ) <EOL> actor . _collect_placement_cb = None <EOL> if done [ <NUM_LIT:0> ] : <EOL> return <EOL> try : <EOL> while True : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { } ) <EOL> node_id = node_iter . next ( ) <EOL> possible_placements . add ( node_id ) <EOL> except dynops . PauseIteration : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , <EOL> { '<STR_LIT>' : actor . _collect_placement_counter , <EOL> '<STR_LIT>' : actor . _collect_placement_last_value , <EOL> '<STR_LIT>' : actor . _collect_placement_counter - actor . _collect_placement_last_value } ) <EOL> delay = <NUM_LIT:0.0> if actor . _collect_placement_counter > actor . _collect_placement_last_value + <NUM_LIT:100> else <NUM_LIT> <EOL> actor . _collect_placement_counter += <NUM_LIT:1> <EOL> actor . _collect_placement_cb = async . DelayedCall ( delay , self . _update_requirements_placements , <EOL> node_iter , actor_id , possible_placements , done = done , <EOL> move = move , cb = cb ) <EOL> return <EOL> except StopIteration : <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { } ) <EOL> done [ <NUM_LIT:0> ] = True <EOL> if move and len ( possible_placements ) > <NUM_LIT:1> : <EOL> possible_placements . discard ( self . node . id ) <EOL> if not possible_placements : <EOL> if cb : <EOL> cb ( status = response . CalvinResponse ( False ) ) <EOL> return <EOL> if self . node . id in possible_placements : <EOL> if cb : <EOL> cb ( status = response . CalvinResponse ( True ) ) <EOL> return <EOL> self . migrate ( actor_id , possible_placements . pop ( ) , callback = cb ) <EOL> _log . analyze ( self . node . id , "<STR_LIT>" , { } ) <EOL> except : <EOL> _log . exception ( "<STR_LIT>" ) <EOL> def migrate ( self , actor_id , node_id , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if actor_id not in self . actors : <EOL> if callback : <EOL> callback ( status = response . CalvinResponse ( False ) ) <EOL> return <EOL> if node_id == self . node . id : <EOL> if callback : <EOL> callback ( status = response . CalvinResponse ( True ) ) <EOL> return <EOL> actor = self . actors [ actor_id ] <EOL> actor . _migrating_to = node_id <EOL> actor . will_migrate ( ) <EOL> actor_type = actor . _type <EOL> ports = actor . connections ( self . node . id ) <EOL> self . node . pm . disconnect ( callback = CalvinCB ( self . _migrate_disconnected , <EOL> actor = actor , <EOL> actor_type = actor_type , <EOL> ports = ports , <EOL> node_id = node_id , <EOL> callback = callback ) , <EOL> actor_id = actor_id ) <EOL> self . node . control . log_actor_migrate ( actor_id , node_id ) <EOL> def _migrate_disconnected ( self , actor , actor_type , ports , node_id , status , callback = None , ** state ) : <EOL> """<STR_LIT>""" <EOL> if status : <EOL> state = actor . state ( ) <EOL> self . destroy ( actor . id ) <EOL> self . node . proto . actor_new ( node_id , callback , actor_type , state , ports ) <EOL> else : <EOL> if callback : <EOL> callback ( status = status ) <EOL> def peernew_to_local_cb ( self , reply , ** kwargs ) : <EOL> if kwargs [ '<STR_LIT>' ] == reply : <EOL> self . node . set_local_reply ( kwargs [ '<STR_LIT>' ] , "<STR_LIT:OK>" ) <EOL> else : <EOL> self . node . set_local_reply ( kwargs [ '<STR_LIT>' ] , reply ) <EOL> def _prev_connections_to_connection_list ( self , prev_connections ) : <EOL> """<STR_LIT>""" <EOL> cl = [ ] <EOL> for in_port_id , out_id in prev_connections [ '<STR_LIT>' ] . iteritems ( ) : <EOL> cl . append ( ( self . node . id , in_port_id , out_id [ <NUM_LIT:0> ] , out_id [ <NUM_LIT:1> ] ) ) <EOL> for out_port_id , in_list in prev_connections [ '<STR_LIT>' ] . iteritems ( ) : <EOL> for in_id in in_list : <EOL> cl . append ( ( self . node . id , out_port_id , in_id [ <NUM_LIT:0> ] , in_id [ <NUM_LIT:1> ] ) ) <EOL> return cl <EOL> def connect ( self , actor_id , connection_list , callback = None ) : <EOL> """<STR_LIT>""" <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> peer_port_ids = [ c [ <NUM_LIT:3> ] for c in connection_list ] <EOL> for node_id , port_id , peer_node_id , peer_port_id in connection_list : <EOL> self . node . pm . connect ( port_id = port_id , <EOL> peer_node_id = peer_node_id , <EOL> peer_port_id = peer_port_id , <EOL> callback = CalvinCB ( self . _actor_connected , <EOL> peer_port_id = peer_port_id , <EOL> actor_id = actor_id , <EOL> peer_port_ids = peer_port_ids , <EOL> _callback = callback ) ) <EOL> def _actor_connected ( self , status , peer_port_id , actor_id , peer_port_ids , _callback , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not status and peer_port_ids : <EOL> if _callback : <EOL> del peer_port_ids [ : ] <EOL> _callback ( status = response . CalvinResponse ( False ) , actor_id = actor_id ) <EOL> if peer_port_id in peer_port_ids : <EOL> peer_port_ids . remove ( peer_port_id ) <EOL> if not peer_port_ids : <EOL> if _callback : <EOL> _callback ( status = response . CalvinResponse ( True ) , actor_id = actor_id ) <EOL> def connections ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> return self . actors [ actor_id ] . connections ( self . node . id ) <EOL> def dump ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> actor = self . actors [ actor_id ] <EOL> _log . debug ( "<STR_LIT>" ) <EOL> _log . debug ( actor ) <EOL> _log . debug ( "<STR_LIT>" ) <EOL> def set_port_property ( self , actor_id , port_type , port_name , port_property , value ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> actor = self . actors [ actor_id ] <EOL> success = actor . set_port_property ( port_type , port_name , port_property , value ) <EOL> return '<STR_LIT:OK>' if success else '<STR_LIT>' <EOL> def get_port_state ( self , actor_id , port_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> actor = self . actors [ actor_id ] <EOL> for port in actor . inports . values ( ) : <EOL> if port . id == port_id : <EOL> return port . fifo . _state ( ) <EOL> for port in actor . outports . values ( ) : <EOL> if port . id == port_id : <EOL> return port . fifo . _state ( ) <EOL> raise Exception ( "<STR_LIT>" % port_id ) <EOL> def actor_type ( self , actor_id ) : <EOL> actor = self . actors . get ( actor_id , None ) <EOL> return actor . _type if actor else '<STR_LIT>' <EOL> def report ( self , actor_id ) : <EOL> if actor_id not in self . actors : <EOL> self . _actor_not_found ( actor_id ) <EOL> return self . actors [ actor_id ] . report ( ) <EOL> def enabled_actors ( self ) : <EOL> return [ actor for actor in self . actors . values ( ) if actor . enabled ( ) ] <EOL> def list_actors ( self ) : <EOL> return self . actors . keys ( ) </s>
