<s> """<STR_LIT>""" <EOL> import time <EOL> import uuid <EOL> import zmq <EOL> from zmq . devices import ThreadDevice <EOL> from zmq . eventloop import ioloop , zmqstream <EOL> from IPython . config . configurable import LoggingConfigurable <EOL> from IPython . utils . traitlets import Set , Instance , CFloat <EOL> from IPython . parallel . util import asbytes <EOL> class Heart ( object ) : <EOL> """<STR_LIT>""" <EOL> device = None <EOL> id = None <EOL> def __init__ ( self , in_addr , out_addr , in_type = zmq . SUB , out_type = zmq . XREQ , heart_id = None ) : <EOL> self . device = ThreadDevice ( zmq . FORWARDER , in_type , out_type ) <EOL> self . device . daemon = True <EOL> self . device . connect_in ( in_addr ) <EOL> self . device . connect_out ( out_addr ) <EOL> if in_type == zmq . SUB : <EOL> self . device . setsockopt_in ( zmq . SUBSCRIBE , b"<STR_LIT>" ) <EOL> if heart_id is None : <EOL> heart_id = uuid . uuid4 ( ) . bytes <EOL> self . device . setsockopt_out ( zmq . IDENTITY , heart_id ) <EOL> self . id = heart_id <EOL> def start ( self ) : <EOL> return self . device . start ( ) <EOL> class HeartMonitor ( LoggingConfigurable ) : <EOL> """<STR_LIT>""" <EOL> period = CFloat ( <NUM_LIT:1000> , config = True , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> ) <EOL> pingstream = Instance ( '<STR_LIT>' ) <EOL> pongstream = Instance ( '<STR_LIT>' ) <EOL> loop = Instance ( '<STR_LIT>' ) <EOL> def _loop_default ( self ) : <EOL> return ioloop . IOLoop . instance ( ) <EOL> hearts = Set ( ) <EOL> responses = Set ( ) <EOL> on_probation = Set ( ) <EOL> last_ping = CFloat ( <NUM_LIT:0> ) <EOL> _new_handlers = Set ( ) <EOL> _failure_handlers = Set ( ) <EOL> lifetime = CFloat ( <NUM_LIT:0> ) <EOL> tic = CFloat ( <NUM_LIT:0> ) <EOL> def __init__ ( self , ** kwargs ) : <EOL> super ( HeartMonitor , self ) . __init__ ( ** kwargs ) <EOL> self . pongstream . on_recv ( self . handle_pong ) <EOL> def start ( self ) : <EOL> self . caller = ioloop . PeriodicCallback ( self . beat , self . period , self . loop ) <EOL> self . caller . start ( ) <EOL> def add_new_heart_handler ( self , handler ) : <EOL> """<STR_LIT>""" <EOL> self . log . debug ( "<STR_LIT>" % handler ) <EOL> self . _new_handlers . add ( handler ) <EOL> def add_heart_failure_handler ( self , handler ) : <EOL> """<STR_LIT>""" <EOL> self . log . debug ( "<STR_LIT>" % handler ) <EOL> self . _failure_handlers . add ( handler ) <EOL> def beat ( self ) : <EOL> self . pongstream . flush ( ) <EOL> self . last_ping = self . lifetime <EOL> toc = time . time ( ) <EOL> self . lifetime += toc - self . tic <EOL> self . tic = toc <EOL> goodhearts = self . hearts . intersection ( self . responses ) <EOL> missed_beats = self . hearts . difference ( goodhearts ) <EOL> heartfailures = self . on_probation . intersection ( missed_beats ) <EOL> newhearts = self . responses . difference ( goodhearts ) <EOL> list ( map ( self . handle_new_heart , newhearts ) ) <EOL> list ( map ( self . handle_heart_failure , heartfailures ) ) <EOL> self . on_probation = missed_beats . intersection ( self . hearts ) <EOL> self . responses = set ( ) <EOL> self . pingstream . send ( asbytes ( str ( self . lifetime ) ) ) <EOL> def handle_new_heart ( self , heart ) : <EOL> if self . _new_handlers : <EOL> for handler in self . _new_handlers : <EOL> handler ( heart ) <EOL> else : <EOL> self . log . info ( "<STR_LIT>" % heart ) <EOL> self . hearts . add ( heart ) <EOL> def handle_heart_failure ( self , heart ) : <EOL> if self . _failure_handlers : <EOL> for handler in self . _failure_handlers : <EOL> try : <EOL> handler ( heart ) <EOL> except Exception as e : <EOL> self . log . error ( "<STR_LIT>" % handler , exc_info = True ) <EOL> pass <EOL> else : <EOL> self . log . info ( "<STR_LIT>" % heart ) <EOL> self . hearts . remove ( heart ) <EOL> def handle_pong ( self , msg ) : <EOL> "<STR_LIT>" <EOL> current = asbytes ( str ( self . lifetime ) ) <EOL> last = asbytes ( str ( self . last_ping ) ) <EOL> if msg [ <NUM_LIT:1> ] == current : <EOL> delta = time . time ( ) - self . tic <EOL> self . responses . add ( msg [ <NUM_LIT:0> ] ) <EOL> elif msg [ <NUM_LIT:1> ] == last : <EOL> delta = time . time ( ) - self . tic + ( self . lifetime - self . last_ping ) <EOL> self . log . warn ( "<STR_LIT>" % ( msg [ <NUM_LIT:0> ] , <NUM_LIT:1000> * delta ) ) <EOL> self . responses . add ( msg [ <NUM_LIT:0> ] ) <EOL> else : <EOL> self . log . warn ( "<STR_LIT>" % <EOL> ( msg [ <NUM_LIT:1> ] , self . lifetime ) ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> loop = ioloop . IOLoop . instance ( ) <EOL> context = zmq . Context ( ) <EOL> pub = context . socket ( zmq . PUB ) <EOL> pub . bind ( '<STR_LIT>' ) <EOL> xrep = context . socket ( zmq . XREP ) <EOL> xrep . bind ( '<STR_LIT>' ) <EOL> outstream = zmqstream . ZMQStream ( pub , loop ) <EOL> instream = zmqstream . ZMQStream ( xrep , loop ) <EOL> hb = HeartMonitor ( loop , outstream , instream ) <EOL> loop . start ( ) </s>
