<s> from collections import defaultdict <EOL> from uuid import uuid4 <EOL> from time import time <EOL> from random import choice <EOL> import operator as op <EOL> from palm . palm import ProtoBase <EOL> from functools import partial <EOL> from diesel import quickstart , Thunk , sleep , log , fork <EOL> from diesel . util . queue import Queue , first <EOL> from diesel . logmod import LOGLVL_DEBUG <EOL> from . convoy_env_palm import MessageResponse , MessageEnvelope <EOL> from . consensus . server import run_server as run_consensus_server <EOL> from . consensus . client import ( ConvoyNameService , ConsensusSet , <EOL> ConvoySetFailed , ConvoySetTimeout , <EOL> ConvoyWaitDone ) <EOL> from . messagenet import ( me , ConvoyService , <EOL> MESSAGE_RES , MESSAGE_OUT , <EOL> host_loop ) <EOL> class ConvoyRemoteException ( object ) : <EOL> def __init__ ( self , s ) : <EOL> self . exc_desc = s <EOL> class ConvoyRemoteNull ( object ) : <EOL> pass <EOL> class ConvoyRemoteResult ( object ) : <EOL> def __init__ ( self , i ) : <EOL> self . i = i <EOL> @ property <EOL> def single ( self ) : <EOL> return self . i [ <NUM_LIT:0> ] <EOL> def __iter__ ( self ) : <EOL> return self . i <EOL> def __len__ ( self ) : <EOL> return len ( self . i ) <EOL> class ConvoyRemoteError ( Exception ) : pass <EOL> class ConvoyTimeoutError ( Exception ) : pass <EOL> class Convoy ( object ) : <EOL> def __init__ ( self ) : <EOL> self . routes = defaultdict ( set ) <EOL> self . local_handlers = { } <EOL> self . enabled_handlers = { } <EOL> self . classes = { } <EOL> self . host_queues = { } <EOL> self . run_nameserver = None <EOL> self . role_messages = defaultdict ( list ) <EOL> self . roles = set ( ) <EOL> self . roles_wanted = set ( ) <EOL> self . roles_owned = set ( ) <EOL> self . role_clocks = { } <EOL> self . role_by_name = { } <EOL> self . incoming = Queue ( ) <EOL> self . pending = { } <EOL> self . rpc_waits = { } <EOL> self . table_changes = Queue ( ) <EOL> def run_with_nameserver ( self , myns , nameservers , * objs ) : <EOL> self . run_nameserver = myns <EOL> self . run ( nameservers , * objs ) <EOL> def run ( self , nameservers , * objs ) : <EOL> nameservers = [ ( h , int ( p ) ) <EOL> for h , p in ( i . split ( '<STR_LIT::>' ) <EOL> for i in nameservers ) ] <EOL> runem = [ ] <EOL> if self . run_nameserver : <EOL> runem . append ( <EOL> Thunk ( lambda : run_consensus_server ( self . run_nameserver , nameservers ) ) ) <EOL> runem . append ( self ) <EOL> handler_functions = dict ( ( v , k ) for k , v in self . local_handlers . iteritems ( ) ) <EOL> final_o = [ ] <EOL> for o in objs : <EOL> if type ( o . __class__ ) is ConvoyRegistrar : <EOL> r = o . __class__ <EOL> self . roles_wanted . add ( r ) <EOL> for m in self . role_messages [ r ] : <EOL> assert m not in self . local_handlers , "<STR_LIT>" <EOL> self . local_handlers [ m ] = getattr ( o , '<STR_LIT>' + m ) <EOL> else : <EOL> final_o . append ( o ) <EOL> self . ns = ConvoyNameService ( nameservers ) <EOL> runem . append ( self . ns ) <EOL> runem . append ( self . deliver ) <EOL> runem . extend ( final_o ) <EOL> runem . append ( ConvoyService ( ) ) <EOL> quickstart ( * runem ) <EOL> def __call__ ( self ) : <EOL> assert me . id <EOL> should_process = self . roles <EOL> rlog = log . sublog ( "<STR_LIT>" , LOGLVL_DEBUG ) <EOL> while True : <EOL> for r in should_process : <EOL> if r in self . roles_wanted : <EOL> resp = self . ns . add ( r . name ( ) , me . id , r . limit ) <EOL> ans = None <EOL> if type ( resp ) == ConsensusSet : <EOL> self . roles_owned . add ( r ) <EOL> ans = resp <EOL> else : <EOL> if r in self . roles_owned : <EOL> self . roles_owned . remove ( r ) <EOL> if resp . set : <EOL> ans = resp . set <EOL> else : <EOL> ans = self . ns . lookup ( r . name ( ) ) <EOL> if ans : <EOL> self . role_clocks [ r . name ( ) ] = ans . clock <EOL> for m in self . role_messages [ r ] : <EOL> self . routes [ m ] = ans . members <EOL> if should_process : <EOL> self . log_resolution_table ( rlog , should_process ) <EOL> self . table_changes . put ( None ) <EOL> wait_result = self . ns . wait ( <NUM_LIT:5> , self . role_clocks ) <EOL> if type ( wait_result ) == ConvoyWaitDone : <EOL> should_process = set ( [ self . role_by_name [ wait_result . key ] ] ) <EOL> else : <EOL> should_process = set ( ) <EOL> self . ns . alive ( ) <EOL> def log_resolution_table ( self , rlog , processed ) : <EOL> rlog . debug ( "<STR_LIT>" ) <EOL> rlog . debug ( "<STR_LIT:U+0020>" ) <EOL> for p in processed : <EOL> rlog . debug ( "<STR_LIT>" % <EOL> ( p . name ( ) , <EOL> '<STR_LIT:U+002CU+0020>' . join ( self . role_messages [ p ] ) ) ) <EOL> if self . role_messages : <EOL> hosts = self . routes [ self . role_messages [ p ] [ <NUM_LIT:0> ] ] <EOL> for h in hosts : <EOL> rlog . debug ( "<STR_LIT>" % ( <EOL> '<STR_LIT:*>' if h == me . id else '<STR_LIT:->' , <EOL> h ) ) <EOL> def register ( self , mod ) : <EOL> for name in dir ( mod ) : <EOL> v = getattr ( mod , name ) <EOL> if type ( v ) is type and issubclass ( v , ProtoBase ) : <EOL> self . classes [ v . __name__ ] = v <EOL> def add_target_role ( self , o ) : <EOL> self . roles . add ( o ) <EOL> self . role_by_name [ o . name ( ) ] = o <EOL> for k , v in o . __dict__ . iteritems ( ) : <EOL> if k . startswith ( "<STR_LIT>" ) and callable ( v ) : <EOL> handler_for = k . split ( "<STR_LIT:_>" , <NUM_LIT:1> ) [ - <NUM_LIT:1> ] <EOL> assert handler_for in self . classes , "<STR_LIT>" <EOL> self . role_messages [ o ] . append ( handler_for ) <EOL> def host_specific_send ( self , host , msg , typ , transport_cb ) : <EOL> if host not in self . host_queues : <EOL> q = Queue ( ) <EOL> fork ( host_loop , host , q ) <EOL> self . host_queues [ host ] = q <EOL> self . host_queues [ host ] . put ( ( msg , typ , transport_cb ) ) <EOL> def local_dispatch ( self , env ) : <EOL> if env . type not in self . classes : <EOL> self . host_specific_send ( env . node_id , <EOL> MessageResponse ( in_response_to = env . req_id , <EOL> result = MessageResponse . REFUSED , <EOL> error_message = "<STR_LIT>" ) , <EOL> MESSAGE_RES , None ) <EOL> elif me . id not in self . routes [ env . type ] : <EOL> self . host_specific_send ( env . node_id , <EOL> MessageResponse ( in_response_to = env . req_id , <EOL> delivered = MessageResponse . REFUSED , <EOL> error_message = "<STR_LIT>" ) , <EOL> MESSAGE_RES , None ) <EOL> else : <EOL> inst = self . classes [ env . type ] ( env . body ) <EOL> r = self . local_handlers [ env . type ] <EOL> sender = ConvoySender ( env ) <EOL> back = MessageResponse ( in_response_to = env . req_id , <EOL> delivered = MessageResponse . ACCEPTED ) <EOL> self . host_specific_send ( env . node_id , back , <EOL> MESSAGE_RES , None ) <EOL> try : <EOL> r ( sender , inst ) <EOL> except Exception , e : <EOL> s = str ( e ) <EOL> back . result = MessageResponse . EXCEPTION <EOL> back . error_message = s <EOL> raise <EOL> else : <EOL> if sender . responses : <EOL> back . result = MessageResponse . RESULT <EOL> back . responses . extend ( sender . responses ) <EOL> else : <EOL> back . result = MessageResponse . NULL <EOL> if env . wants_result : <EOL> back . delivered = MessageResponse . FINISHED <EOL> self . host_specific_send ( env . node_id , back , <EOL> MESSAGE_RES , None ) <EOL> def local_response ( self , result ) : <EOL> id = result . in_response_to <EOL> if result . delivered == MessageResponse . REFUSED : <EOL> self . retry ( id ) <EOL> elif result . delivered == MessageResponse . ACCEPTED : <EOL> if id in self . pending : <EOL> del self . pending [ id ] <EOL> elif result . delivered == MessageResponse . FINISHED : <EOL> if id in self . rpc_waits : <EOL> q = self . rpc_waits . pop ( id ) <EOL> if result . result == MessageResponse . EXCEPTION : <EOL> resp = ConvoyRemoteException ( result . error_message ) <EOL> elif result . result == MessageResponse . NULL : <EOL> resp = ConvoyRemoteNull ( ) <EOL> elif result . result == MessageResponse . RESULT : <EOL> res = [ self . classes [ m . type ] ( m . body ) <EOL> for m in result . responses ] <EOL> resp = ConvoyRemoteResult ( res ) <EOL> else : <EOL> assert <NUM_LIT:0> <EOL> q . put ( resp ) <EOL> def send ( self , m , timeout = <NUM_LIT:10> ) : <EOL> self . incoming . put ( Delivery ( m , timeout ) ) <EOL> def broadcast ( self , m ) : <EOL> pass <EOL> def rpc ( self , m , timeout = <NUM_LIT:10> ) : <EOL> q = Queue ( ) <EOL> self . incoming . put ( Delivery ( m , timeout , rqueue = q ) ) <EOL> ev , res = first ( sleep = timeout , waits = [ q ] ) <EOL> if ev == q : <EOL> if res == ConvoyRemoteException : <EOL> raise ConvoyRemoteError ( res . exc_desc ) <EOL> if res == ConvoyRemoteNull : <EOL> return None <EOL> return res <EOL> else : <EOL> raise ConvoyTimeoutError ( "<STR_LIT>" + <EOL> ( "<STR_LIT>" % timeout ) ) <EOL> def retry ( self , id ) : <EOL> if id in self . pending : <EOL> next = self . pending . pop ( id ) <EOL> self . incoming . put ( next ) <EOL> def deliver ( self ) : <EOL> deferred = [ ] <EOL> srg = self . routes . get <EOL> empty = set ( ) <EOL> sorter = op . attrgetter ( "<STR_LIT>" ) <EOL> while True : <EOL> wait = ( <NUM_LIT:1.0> if not deferred else <EOL> deferred [ - <NUM_LIT:1> ] . remaining ) <EOL> r , next = first ( waits = [ self . incoming , <EOL> self . table_changes ] , sleep = wait ) <EOL> if r == self . incoming : <EOL> if next . rqueue : <EOL> self . rpc_waits [ next . id ] = next . rqueue <EOL> hosts = srg ( next . target , empty ) <EOL> potentials = hosts - next . hosts_tried <EOL> if not potentials : <EOL> next . reschedule ( ) <EOL> deferred . append ( next ) <EOL> else : <EOL> host = choice ( list ( potentials ) ) <EOL> next . hosts_tried . add ( host ) <EOL> self . pending [ next . id ] = next <EOL> self . host_specific_send ( host , next . env , <EOL> MESSAGE_OUT , <EOL> partial ( self . retry , next . id ) ) <EOL> deferred . sort ( key = sorter , reverse = True ) <EOL> t = time ( ) <EOL> while deferred and deferred [ - <NUM_LIT:1> ] . due ( t ) : <EOL> i = deferred . pop ( ) <EOL> if not i . expired ( t ) : <EOL> self . incoming . put ( i ) <EOL> class Delivery ( object ) : <EOL> def __init__ ( self , m , timeout , rqueue = None , broadcast = False ) : <EOL> self . id = str ( uuid4 ( ) ) <EOL> self . target = m . __class__ . __name__ <EOL> self . timeout = time ( ) + timeout <EOL> self . rqueue = rqueue <EOL> self . hosts_tried = set ( ) <EOL> self . reschedule_at = None <EOL> self . reschedule_interval = <NUM_LIT> <EOL> self . m = m <EOL> self . broadcast = broadcast <EOL> self . env = MessageEnvelope ( <EOL> body = m . dumps ( ) , <EOL> type = self . target , <EOL> req_id = self . id , <EOL> node_id = me . id , <EOL> wants_result = bool ( rqueue ) ) <EOL> def due ( self , t ) : <EOL> return t >= self . reschedule_at <EOL> def reschedule ( self ) : <EOL> self . reschedule_at = min ( <EOL> time ( ) + self . reschedule_interval , <EOL> self . timeout ) <EOL> self . reschedule_interval *= <NUM_LIT:2> <EOL> self . hosts_tried = set ( ) <EOL> @ property <EOL> def remaining ( self ) : <EOL> return max ( <NUM_LIT:0> , self . reschedule_at <EOL> - time ( ) ) <EOL> def expired ( self , t ) : <EOL> return t >= self . timeout <EOL> class ConvoySender ( object ) : <EOL> def __init__ ( self , env ) : <EOL> self . from_host = env . node_id <EOL> self . type = env . type <EOL> self . req_id = env . req_id <EOL> self . responses = [ ] <EOL> def respond ( self , m ) : <EOL> env = MessageEnvelope ( body = m . dumps ( ) , <EOL> type = m . __class__ . __name__ , <EOL> req_id = '<STR_LIT>' , <EOL> node_id = '<STR_LIT>' , <EOL> wants_result = False ) <EOL> self . responses . append ( env ) <EOL> convoy = Convoy ( ) <EOL> class ConvoyRegistrar ( type ) : <EOL> def __new__ ( * args ) : <EOL> t = type . __new__ ( * args ) <EOL> if t . __name__ != '<STR_LIT>' : <EOL> convoy . add_target_role ( t ) <EOL> return t <EOL> class ConvoyRole ( object ) : <EOL> __metaclass__ = ConvoyRegistrar <EOL> limit = <NUM_LIT:0> <EOL> @ classmethod <EOL> def name ( cls ) : <EOL> return cls . __name__ </s>
