<s> from __future__ import print_function <EOL> import sys <EOL> import random <EOL> from types import GeneratorType <EOL> from spinoff . util . python import enumrange <EOL> from spinoff . util . logging import dbg <EOL> ( <EOL> Bind , Connect , Disconnect , SigDisconnect , Send , Ping , <EOL> RelaySigNew , RelayConnect , RelaySigConnected , RelaySend , RelayForward , RelaySigNodeDown , RelayNvm , <EOL> Receive , SendFailed , NodeDown , NextBeat <EOL> ) = enumrange ( <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ) <EOL> IN , OUT = enumrange ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> BIG_BANG_T = - sys . maxint <EOL> def nid2addr ( nid ) : <EOL> return nid . rsplit ( '<STR_LIT:|>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> def FLUSH ( self , nid ) : <EOL> for msg_h in self . queues . pop ( nid , [ ] ) : <EOL> yield SendFailed , msg_h <EOL> def NODEDOWN ( self , nid ) : <EOL> yield FLUSH ( self , nid ) <EOL> yield NodeDown , nid <EOL> for x in [ self . last_seen , self . last_sent , self . versions ] : <EOL> x . pop ( nid , None ) <EOL> def RELAY_NODEDOWN_CHECKS ( self , nid ) : <EOL> for relayee_nid in self . rl_relayers . pop ( nid , set ( ) ) : <EOL> relayees_relayers = self . rl_relayees [ relayee_nid ] <EOL> relayees_relayers . remove ( nid ) <EOL> if not relayees_relayers : <EOL> del self . rl_relayees [ relayee_nid ] <EOL> for relayer_nid in self . rl_relayees . pop ( nid , set ( ) ) : <EOL> relayers_relayees = self . rl_relayers [ relayer_nid ] <EOL> relayers_relayees . remove ( nid ) <EOL> if not relayers_relayees : <EOL> del self . rl_relayers [ relayer_nid ] <EOL> yield RelaySigNodeDown , IN , relayer_nid , nid <EOL> class HubLogic ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , nid , heartbeat_interval , heartbeat_max_silence , is_relay = False ) : <EOL> self . nid = nid <EOL> self . heartbeat_interval = heartbeat_interval <EOL> self . heartbeat_max_silence = heartbeat_max_silence <EOL> self . is_relay = is_relay <EOL> self . channels_in = set ( ) <EOL> self . channels_out = set ( ) <EOL> self . last_seen = { } <EOL> self . last_sent = { } <EOL> self . versions = { } <EOL> self . queues = { } <EOL> self . rl_relayees = { } <EOL> self . rl_relayers = { } <EOL> self . cl_avail_relays = { } <EOL> self . cl_relayees = { } <EOL> self . version = - <NUM_LIT:1> <EOL> def start ( self ) : <EOL> port = self . nid . split ( '<STR_LIT::>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> yield Bind , nid2addr ( '<STR_LIT>' + port ) <EOL> yield NextBeat , self . heartbeat_interval <EOL> def send_message ( self , rcpt_nid , msg_h , t ) : <EOL> if rcpt_nid in self . channels_in : <EOL> yield Send , IN , rcpt_nid , self . _next_version ( ) , msg_h <EOL> elif rcpt_nid in self . channels_out : <EOL> if rcpt_nid not in self . queues : <EOL> yield Send , OUT , rcpt_nid , self . _next_version ( ) , msg_h <EOL> else : <EOL> self . queues [ rcpt_nid ] . append ( msg_h ) <EOL> elif rcpt_nid in self . queues : <EOL> self . queues [ rcpt_nid ] . append ( msg_h ) <EOL> elif rcpt_nid in self . cl_relayees : <EOL> relay_nid = self . cl_relayees [ rcpt_nid ] <EOL> yield RelaySend , ( IN if relay_nid in self . channels_in else OUT ) , relay_nid , rcpt_nid , msg_h <EOL> else : <EOL> self . last_sent [ rcpt_nid ] = t <EOL> self . channels_out . add ( rcpt_nid ) <EOL> self . last_seen [ rcpt_nid ] = t <EOL> self . queues . setdefault ( rcpt_nid , [ ] ) . append ( msg_h ) <EOL> yield Connect , nid2addr ( rcpt_nid ) <EOL> yield Ping , OUT , rcpt_nid , self . _next_version ( ) <EOL> if self . is_relay : <EOL> yield RelaySigNew , OUT , rcpt_nid <EOL> def sig_disconnect_received ( self , sender_nid ) : <EOL> if sender_nid in self . channels_in or sender_nid in self . channels_out : <EOL> if sender_nid in self . channels_in : <EOL> self . channels_in . remove ( sender_nid ) <EOL> if sender_nid in self . channels_out : <EOL> self . channels_out . remove ( sender_nid ) <EOL> yield Disconnect , nid2addr ( sender_nid ) <EOL> else : <EOL> return <EOL> if sender_nid in self . cl_avail_relays : <EOL> yield self . _handle_relay_down ( sender_nid ) <EOL> else : <EOL> yield RELAY_NODEDOWN_CHECKS ( self , sender_nid ) <EOL> yield NODEDOWN ( self , sender_nid ) <EOL> def ping_received ( self , on_sock , sender_nid , version , t ) : <EOL> if on_sock == OUT and sender_nid not in self . channels_out : <EOL> return <EOL> self . last_seen [ sender_nid ] = t <EOL> if on_sock == IN and sender_nid not in self . channels_in : <EOL> self . channels_in . add ( sender_nid ) <EOL> if self . is_relay : <EOL> yield RelaySigNew , IN , sender_nid <EOL> elif sender_nid in self . cl_relayees : <EOL> relay_nid = self . cl_relayees . pop ( sender_nid ) <EOL> self . cl_avail_relays [ relay_nid ] . remove ( sender_nid ) <EOL> yield RelayNvm , ( IN if relay_nid in self . channels_in else OUT ) , relay_nid , sender_nid <EOL> inout = ( IN if sender_nid in self . channels_in else OUT ) <EOL> if self . _needs_ping ( sender_nid , t ) : <EOL> yield Ping , inout , sender_nid , self . _next_version ( ) <EOL> if sender_nid in self . queues : <EOL> for msg_h in self . queues . pop ( sender_nid ) : <EOL> yield Send , inout , sender_nid , self . _next_version ( ) , msg_h <EOL> else : <EOL> if not ( version > self . versions . get ( sender_nid , - <NUM_LIT:1> ) ) : <EOL> self . versions [ sender_nid ] = version <EOL> assert sender_nid not in self . queues <EOL> if sender_nid in self . cl_avail_relays : <EOL> self . _handle_relay_down ( sender_nid ) <EOL> yield NodeDown , sender_nid <EOL> self . versions [ sender_nid ] = version <EOL> def message_received ( self , on_sock , sender_nid , version , msg_body_bytes , t ) : <EOL> yield self . ping_received ( on_sock , sender_nid , version , t ) <EOL> if msg_body_bytes : <EOL> yield Receive , sender_nid , msg_body_bytes <EOL> def heartbeat ( self , t ) : <EOL> t_gone = t - self . heartbeat_max_silence <EOL> for nid in ( self . channels_in | self . channels_out ) : <EOL> if self . last_seen [ nid ] <= t_gone : <EOL> if nid in self . channels_out : <EOL> self . channels_out . remove ( nid ) <EOL> if nid in self . channels_in : <EOL> self . channels_in . remove ( nid ) <EOL> yield Disconnect , nid2addr ( nid ) <EOL> yield RELAY_NODEDOWN_CHECKS ( self , nid ) <EOL> if self . is_relay or not self . cl_avail_relays : <EOL> yield NODEDOWN ( self , nid ) <EOL> elif nid in self . cl_avail_relays : <EOL> self . _handle_relay_down ( nid ) <EOL> yield NODEDOWN ( self , nid ) <EOL> else : <EOL> relay_nid = random . choice ( self . cl_avail_relays . keys ( ) ) <EOL> self . cl_avail_relays [ relay_nid ] . add ( nid ) <EOL> self . cl_relayees [ nid ] = relay_nid <EOL> yield RelayConnect , ( IN if relay_nid in self . channels_in else OUT ) , relay_nid , nid <EOL> else : <EOL> if self . _needs_ping ( nid , t ) : <EOL> yield Ping , ( IN if nid in self . channels_in else OUT ) , nid , self . _next_version ( ) <EOL> yield NextBeat , self . heartbeat_interval <EOL> def new_relay_received ( self , nid ) : <EOL> self . cl_avail_relays [ nid ] = set ( ) <EOL> def relay_connect_received ( self , on_sock , relayer_nid , relayee_nid ) : <EOL> if on_sock == IN and relayer_nid in self . channels_in or on_sock == OUT and relayer_nid in self . channels_out : <EOL> if relayee_nid in self . channels_in : <EOL> self . rl_relayees . setdefault ( relayee_nid , set ( ) ) . add ( relayer_nid ) <EOL> self . rl_relayers . setdefault ( relayer_nid , set ( ) ) . add ( relayee_nid ) <EOL> yield RelaySigConnected , on_sock , relayer_nid , relayee_nid <EOL> else : <EOL> yield RelaySigNodeDown , on_sock , relayer_nid , relayee_nid <EOL> def relay_connected_received ( self , relayee_nid ) : <EOL> if relayee_nid in self . cl_relayees : <EOL> relay_nid = self . cl_relayees [ relayee_nid ] <EOL> for msg_h in self . queues . pop ( relayee_nid , [ ] ) : <EOL> yield RelaySend , ( IN if relay_nid in self . channels_in else OUT ) , relay_nid , relayee_nid , msg_h <EOL> def relay_nodedown_received ( self , relay_nid , relayee_nid ) : <EOL> if relayee_nid in self . cl_avail_relays . get ( relay_nid , set ( ) ) : <EOL> self . cl_avail_relays [ relay_nid ] . remove ( relayee_nid ) <EOL> del self . cl_relayees [ relayee_nid ] <EOL> yield NODEDOWN ( self , relayee_nid ) <EOL> def relay_send_received ( self , relayer_nid , relayee_nid , relayed_bytes ) : <EOL> if relayee_nid not in self . channels_in : <EOL> return <EOL> if relayee_nid not in self . rl_relayers . get ( relayer_nid , set ( ) ) : <EOL> self . rl_relayers . setdefault ( relayer_nid , set ( ) ) . add ( relayee_nid ) <EOL> self . rl_relayees . setdefault ( relayee_nid , set ( ) ) . add ( relayer_nid ) <EOL> yield RelayForward , IN , relayee_nid , relayer_nid , relayed_bytes <EOL> def relay_forwarded_received ( self , actual_sender_nid , relayed_bytes ) : <EOL> yield Receive , actual_sender_nid , relayed_bytes <EOL> def relay_nvm_received ( self , sender_nid , relayee_nid ) : <EOL> self . rl_relayees . get ( relayee_nid , set ( ) ) . discard ( sender_nid ) <EOL> self . rl_relayers . get ( sender_nid , set ( ) ) . discard ( relayee_nid ) <EOL> def ensure_connected ( self , nid , t ) : <EOL> if nid not in self . channels_in and nid not in self . channels_out and nid not in self . cl_relayees : <EOL> self . last_seen [ nid ] = t <EOL> self . channels_out . add ( nid ) <EOL> self . queues [ nid ] = [ ] <EOL> yield Connect , nid2addr ( nid ) <EOL> self . last_sent [ nid ] = t <EOL> yield Ping , OUT , nid , self . _next_version ( ) <EOL> if self . is_relay : <EOL> yield RelaySigNew , OUT , nid <EOL> def shutdown ( self ) : <EOL> for nid in ( self . channels_in | self . channels_out ) : <EOL> yield SigDisconnect , ( IN if nid in self . channels_in else OUT ) , nid <EOL> yield FLUSH ( self , nid ) <EOL> self . channels_in = self . channels_out = None <EOL> def _handle_relay_down ( self , relay_nid ) : <EOL> for relayee_nid in self . cl_avail_relays . pop ( relay_nid ) : <EOL> del self . cl_relayees [ relayee_nid ] <EOL> yield NODEDOWN ( self , relayee_nid ) <EOL> def _needs_ping ( self , nid , t ) : <EOL> ret = self . last_sent . get ( nid , BIG_BANG_T ) <= t - self . heartbeat_interval / <NUM_LIT> <EOL> if ret : <EOL> self . last_sent [ nid ] = t <EOL> return ret <EOL> def _next_version ( self ) : <EOL> self . version += <NUM_LIT:1> <EOL> return self . version <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" <EOL> def flatten ( gen ) : <EOL> for x in gen : <EOL> if isinstance ( x , GeneratorType ) : <EOL> for y in flatten ( x ) : <EOL> yield y <EOL> else : <EOL> yield x </s>
