<s> __author__ = '<STR_LIT>' <EOL> from pyon . util . log import log <EOL> from pyon . core . thread import PyonThreadManager , PyonThread , ThreadManager , PyonThreadTraceback , PyonHeartbeatError <EOL> from pyon . ion . service import BaseService <EOL> from gevent . event import Event , waitall , AsyncResult <EOL> from gevent . queue import Queue <EOL> from gevent import greenlet , Timeout <EOL> from pyon . util . async import spawn <EOL> from pyon . core . exception import IonException , ContainerError <EOL> from pyon . core . exception import Timeout as IonTimeout <EOL> from pyon . util . containers import get_ion_ts , get_ion_ts_millis <EOL> from pyon . core . bootstrap import CFG <EOL> import threading <EOL> import traceback <EOL> STAT_INTERVAL_LENGTH = <NUM_LIT> <EOL> class OperationInterruptedException ( BaseException ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class IonProcessError ( StandardError ) : <EOL> pass <EOL> class IonProcessThread ( PyonThread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , target = None , listeners = None , name = None , service = None , cleanup_method = None , heartbeat_secs = <NUM_LIT:10> , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> self . _startup_listeners = listeners or [ ] <EOL> self . listeners = [ ] <EOL> self . _listener_map = { } <EOL> self . name = name <EOL> self . service = service <EOL> self . _cleanup_method = cleanup_method <EOL> self . thread_manager = ThreadManager ( failure_notify_callback = self . _child_failed ) <EOL> self . _dead_children = [ ] <EOL> self . _ctrl_thread = None <EOL> self . _ctrl_queue = Queue ( ) <EOL> self . _ready_control = Event ( ) <EOL> self . _errors = [ ] <EOL> self . _ctrl_current = None <EOL> self . _start_time = None <EOL> self . _proc_time = <NUM_LIT:0> <EOL> self . _proc_time_prior = <NUM_LIT:0> <EOL> self . _proc_time_prior2 = <NUM_LIT:0> <EOL> self . _proc_interval_num = <NUM_LIT:0> <EOL> self . _heartbeat_secs = heartbeat_secs <EOL> self . _heartbeat_stack = None <EOL> self . _heartbeat_time = None <EOL> self . _heartbeat_op = None <EOL> self . _heartbeat_count = <NUM_LIT:0> <EOL> PyonThread . __init__ ( self , target = target , ** kwargs ) <EOL> def heartbeat ( self ) : <EOL> """<STR_LIT>""" <EOL> listeners_ok = True <EOL> for l in self . listeners : <EOL> if not ( l in self . _listener_map and not self . _listener_map [ l ] . proc . dead and l . get_ready_event ( ) . is_set ( ) ) : <EOL> listeners_ok = False <EOL> ctrl_thread_ok = self . _ctrl_thread . running <EOL> heartbeat_ok = True <EOL> if self . _ctrl_current is not None : <EOL> st = traceback . extract_stack ( self . _ctrl_thread . proc . gr_frame ) <EOL> if self . _ctrl_current == self . _heartbeat_op : <EOL> if st == self . _heartbeat_stack : <EOL> self . _heartbeat_count += <NUM_LIT:1> <EOL> if self . _heartbeat_count > CFG . get_safe ( '<STR_LIT>' , <NUM_LIT:30> ) or get_ion_ts_millis ( ) - int ( self . _heartbeat_time ) >= CFG . get_safe ( '<STR_LIT>' , <NUM_LIT:30> ) * <NUM_LIT:1000> : <EOL> heartbeat_ok = False <EOL> else : <EOL> self . _heartbeat_count = <NUM_LIT:1> <EOL> self . _heartbeat_stack = st <EOL> self . _heartbeat_time = get_ion_ts ( ) <EOL> else : <EOL> self . _heartbeat_op = self . _ctrl_current <EOL> self . _heartbeat_count = <NUM_LIT:1> <EOL> self . _heartbeat_time = get_ion_ts ( ) <EOL> self . _heartbeat_stack = st <EOL> else : <EOL> self . _heartbeat_op = None <EOL> self . _heartbeat_count = <NUM_LIT:0> <EOL> return ( listeners_ok , ctrl_thread_ok , heartbeat_ok ) <EOL> @ property <EOL> def time_stats ( self ) : <EOL> """<STR_LIT>""" <EOL> now = get_ion_ts_millis ( ) <EOL> running_time = now - self . _start_time <EOL> idle_time = running_time - self . _proc_time <EOL> cur_interval = now / STAT_INTERVAL_LENGTH <EOL> now_since_prior = now - ( cur_interval - <NUM_LIT:1> ) * STAT_INTERVAL_LENGTH <EOL> if cur_interval == self . _proc_interval_num : <EOL> proc_time_since_prior = self . _proc_time - self . _proc_time_prior2 <EOL> elif cur_interval - <NUM_LIT:1> == self . _proc_interval_num : <EOL> proc_time_since_prior = self . _proc_time - self . _proc_time_prior <EOL> else : <EOL> proc_time_since_prior = <NUM_LIT:0> <EOL> return ( running_time , idle_time , self . _proc_time , now_since_prior , proc_time_since_prior ) <EOL> def _child_failed ( self , child ) : <EOL> """<STR_LIT>""" <EOL> for x in self . thread_manager . children : <EOL> if x . proc == child : <EOL> self . thread_manager . children . remove ( x ) <EOL> break <EOL> self . _dead_children . append ( child ) <EOL> self . proc . kill ( child . exception ) <EOL> def add_endpoint ( self , listener ) : <EOL> """<STR_LIT>""" <EOL> if self . proc : <EOL> listener . routing_call = self . _routing_call <EOL> if self . name : <EOL> svc_name = "<STR_LIT>" <EOL> if self . service is not None and hasattr ( self . service , '<STR_LIT:name>' ) : <EOL> svc_name = self . service . name <EOL> listen_thread_name = "<STR_LIT>" % ( svc_name , self . name , len ( self . listeners ) + <NUM_LIT:1> ) <EOL> else : <EOL> listen_thread_name = "<STR_LIT>" % ( len ( self . listeners ) + <NUM_LIT:1> ) <EOL> gl = self . thread_manager . spawn ( listener . listen , thread_name = listen_thread_name ) <EOL> gl . proc . _glname = "<STR_LIT>" % listen_thread_name <EOL> self . _listener_map [ listener ] = gl <EOL> self . listeners . append ( listener ) <EOL> else : <EOL> self . _startup_listeners . append ( listener ) <EOL> def remove_endpoint ( self , listener ) : <EOL> """<STR_LIT>""" <EOL> if listener in self . listeners : <EOL> self . listeners . remove ( listener ) <EOL> return self . _listener_map . pop ( listener ) <EOL> elif listener in self . _startup_listeners : <EOL> self . _startup_listeners . remove ( listener ) <EOL> return None <EOL> else : <EOL> raise IonProcessError ( "<STR_LIT>" % listener ) <EOL> def target ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . name : <EOL> threading . current_thread ( ) . name = "<STR_LIT>" % self . name <EOL> self . _start_time = get_ion_ts_millis ( ) <EOL> self . _proc_interval_num = self . _start_time / STAT_INTERVAL_LENGTH <EOL> self . _ctrl_thread = self . thread_manager . spawn ( self . _control_flow ) <EOL> self . _ctrl_thread . proc . _glname = "<STR_LIT>" % self . name <EOL> while not self . _ctrl_thread . ev_exit . wait ( timeout = self . _heartbeat_secs ) : <EOL> hbst = self . heartbeat ( ) <EOL> if not all ( hbst ) : <EOL> log . warn ( "<STR_LIT>" , self , hbst ) <EOL> if self . _heartbeat_stack is not None : <EOL> stack_out = "<STR_LIT>" . join ( traceback . format_list ( self . _heartbeat_stack ) ) <EOL> else : <EOL> stack_out = "<STR_LIT>" <EOL> log . warn ( "<STR_LIT>" , hbst , stack_out ) <EOL> self . _ctrl_thread . join ( ) <EOL> def _routing_call ( self , call , context , * callargs , ** callkwargs ) : <EOL> """<STR_LIT>""" <EOL> ar = AsyncResult ( ) <EOL> if len ( callargs ) == <NUM_LIT:0> and len ( callkwargs ) == <NUM_LIT:0> : <EOL> log . trace ( "<STR_LIT>" , call ) <EOL> self . _ctrl_queue . put ( ( greenlet . getcurrent ( ) , ar , call , callargs , callkwargs , context ) ) <EOL> return ar <EOL> def has_pending_call ( self , ar ) : <EOL> """<STR_LIT>""" <EOL> for _ , qar , _ , _ , _ , _ in self . _ctrl_queue . queue : <EOL> if qar == ar : <EOL> return True <EOL> return False <EOL> def _cancel_pending_call ( self , ar ) : <EOL> """<STR_LIT>""" <EOL> if self . has_pending_call ( ar ) : <EOL> ar . set ( False ) <EOL> return True <EOL> return False <EOL> def _interrupt_control_thread ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _ctrl_thread . proc . kill ( exception = OperationInterruptedException , block = False ) <EOL> def cancel_or_abort_call ( self , ar ) : <EOL> """<STR_LIT>""" <EOL> if not self . _cancel_pending_call ( ar ) and not ar . ready ( ) : <EOL> self . _interrupt_control_thread ( ) <EOL> def _control_flow ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . name : <EOL> svc_name = "<STR_LIT>" <EOL> if self . service is not None and hasattr ( self . service , '<STR_LIT:name>' ) : <EOL> svc_name = self . service . name <EOL> threading . current_thread ( ) . name = "<STR_LIT>" % ( svc_name , self . name ) <EOL> self . _ready_control . set ( ) <EOL> for calltuple in self . _ctrl_queue : <EOL> calling_gl , ar , call , callargs , callkwargs , context = calltuple <EOL> res = None <EOL> start_proc_time = get_ion_ts_millis ( ) <EOL> self . _record_proc_time ( start_proc_time ) <EOL> if context is not None and '<STR_LIT>' in context : <EOL> if start_proc_time >= int ( context [ '<STR_LIT>' ] ) : <EOL> log . info ( "<STR_LIT>" ) <EOL> e = IonTimeout ( "<STR_LIT>" % ( context [ '<STR_LIT>' ] , start_proc_time ) ) <EOL> calling_gl . kill ( exception = e , block = False ) <EOL> continue <EOL> if ar . ready ( ) : <EOL> log . info ( "<STR_LIT>" ) <EOL> continue <EOL> try : <EOL> with self . service . push_context ( context ) : <EOL> with self . service . container . context . push_context ( context ) : <EOL> self . _ctrl_current = ar <EOL> res = call ( * callargs , ** callkwargs ) <EOL> except OperationInterruptedException : <EOL> log . debug ( "<STR_LIT>" ) <EOL> pass <EOL> except Exception as e : <EOL> exc = PyonThreadTraceback ( "<STR_LIT>" % ( call , callargs , callkwargs , context , traceback . format_exc ( ) ) ) <EOL> e . args = e . args + ( exc , ) <EOL> if isinstance ( e , ( TypeError , IonException ) ) : <EOL> calling_gl . kill ( exception = e , block = False ) <EOL> else : <EOL> self . _errors . append ( ( call , callargs , callkwargs , context , e , exc ) ) <EOL> log . warn ( exc ) <EOL> log . warn ( "<STR_LIT>" ) <EOL> calling_gl . kill ( exception = ContainerError ( str ( exc ) ) , block = False ) <EOL> finally : <EOL> self . _compute_proc_stats ( start_proc_time ) <EOL> self . _ctrl_current = None <EOL> ar . set ( res ) <EOL> def _record_proc_time ( self , cur_time ) : <EOL> """<STR_LIT>""" <EOL> cur_interval = cur_time / STAT_INTERVAL_LENGTH <EOL> if cur_interval == self . _proc_interval_num : <EOL> pass <EOL> elif cur_interval - <NUM_LIT:1> == self . _proc_interval_num : <EOL> self . _proc_interval_num = cur_interval <EOL> self . _proc_time_prior2 = self . _proc_time_prior <EOL> self . _proc_time_prior = self . _proc_time <EOL> elif cur_interval - <NUM_LIT:1> > self . _proc_interval_num : <EOL> self . _proc_interval_num = cur_interval <EOL> self . _proc_time_prior2 = self . _proc_time <EOL> self . _proc_time_prior = self . _proc_time <EOL> def _compute_proc_stats ( self , start_proc_time ) : <EOL> cur_time = get_ion_ts_millis ( ) <EOL> self . _record_proc_time ( cur_time ) <EOL> proc_time = cur_time - start_proc_time <EOL> self . _proc_time += proc_time <EOL> def start_listeners ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . thread_manager . _failure_notify_callback = None <EOL> for listener in self . _startup_listeners : <EOL> self . add_endpoint ( listener ) <EOL> with Timeout ( seconds = CFG . get_safe ( '<STR_LIT>' , <NUM_LIT:30> ) ) : <EOL> waitall ( [ x . get_ready_event ( ) for x in self . listeners ] ) <EOL> except Timeout : <EOL> for listener , proc in self . _listener_map . iteritems ( ) : <EOL> if proc . proc . dead : <EOL> log . info ( "<STR_LIT>" , listener ) <EOL> self . listeners . remove ( listener ) <EOL> self . thread_manager . children . remove ( proc ) <EOL> raise IonProcessError ( "<STR_LIT>" ) <EOL> finally : <EOL> self . thread_manager . _failure_notify_callback = self . _child_failed <EOL> def _notify_stop ( self ) : <EOL> """<STR_LIT>""" <EOL> for listener in self . listeners : <EOL> try : <EOL> listener . close ( ) <EOL> except Exception as ex : <EOL> tb = traceback . format_exc ( ) <EOL> log . warn ( "<STR_LIT>" , ex , tb ) <EOL> self . _ctrl_queue . put ( StopIteration ) <EOL> self . thread_manager . wait_children ( <NUM_LIT:30> ) <EOL> PyonThread . _notify_stop ( self ) <EOL> if self . _cleanup_method is not None : <EOL> try : <EOL> self . _cleanup_method ( self ) <EOL> except Exception as ex : <EOL> log . warn ( "<STR_LIT>" , ex , traceback . format_exc ( ) ) <EOL> def get_ready_event ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _ready_control <EOL> class IonProcessThreadManager ( PyonThreadManager ) : <EOL> def _create_thread ( self , target = None , ** kwargs ) : <EOL> return IonProcessThread ( target = target , heartbeat_secs = self . heartbeat_secs , ** kwargs ) <EOL> class StandaloneProcess ( BaseService ) : <EOL> """<STR_LIT>""" <EOL> process_type = "<STR_LIT>" <EOL> class SimpleProcess ( BaseService ) : <EOL> """<STR_LIT>""" <EOL> process_type = "<STR_LIT>" <EOL> class ImmediateProcess ( BaseService ) : <EOL> """<STR_LIT>""" <EOL> process_type = "<STR_LIT>" <EOL> def get_ion_actor_id ( process ) : <EOL> """<STR_LIT>""" <EOL> ion_actor_id = None <EOL> if process : <EOL> ctx = process . get_context ( ) <EOL> ion_actor_id = ctx . get ( '<STR_LIT>' , None ) if ctx else None <EOL> return ion_actor_id </s>
