<s> '''<STR_LIT>''' <EOL> import argparse <EOL> import os , sys <EOL> import random , struct <EOL> import cPickle as pickle <EOL> import time , socket , select <EOL> from Queue import Queue <EOL> from threading import Thread , RLock , Lock , Condition , Timer , Semaphore <EOL> from concoord . enums import * <EOL> from concoord . nameserver import Nameserver <EOL> from concoord . exception import ConnectionError <EOL> from concoord . utils import * <EOL> from concoord . message import * <EOL> from concoord . pack import * <EOL> from concoord . pvalue import PValueSet <EOL> from concoord . connection import ConnectionPool , Connection , SelfConnection <EOL> try : <EOL> import dns . resolver , dns . exception <EOL> except : <EOL> print ( "<STR_LIT>" ) <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT:port>" , type = int , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT>" , default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT>" , default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store>" , dest = "<STR_LIT>" , default = '<STR_LIT>' , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store_true>" , dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store_true>" , dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_argument ( "<STR_LIT>" , "<STR_LIT>" , action = "<STR_LIT:store_true>" , dest = "<STR_LIT>" , default = False , <EOL> help = "<STR_LIT>" ) <EOL> args = parser . parse_args ( ) <EOL> class Node ( ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> addr = args . addr , <EOL> port = args . port , <EOL> givenbootstraplist = args . bootstrap , <EOL> debugoption = args . debug , <EOL> objectname = args . objectname , <EOL> logger = args . logger , <EOL> writetodisk = args . writetodisk ) : <EOL> self . addr = addr if addr else findOwnIP ( ) <EOL> self . port = port <EOL> self . debug = debugoption <EOL> self . durable = writetodisk <EOL> self . isnameserver = args . domain != '<STR_LIT>' <EOL> self . domain = args . domain <EOL> self . useroute53 = args . route53 <EOL> if objectname == '<STR_LIT>' : <EOL> parser . print_help ( ) <EOL> self . _graceexit ( <NUM_LIT:1> ) <EOL> self . objectname = objectname <EOL> self . receivedmessages_semaphore = Semaphore ( <NUM_LIT:0> ) <EOL> self . receivedmessages = [ ] <EOL> self . lock = Lock ( ) <EOL> self . socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> self . socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> self . socket . setsockopt ( socket . IPPROTO_TCP , socket . TCP_NODELAY , <NUM_LIT:1> ) <EOL> self . socket . setblocking ( <NUM_LIT:0> ) <EOL> if self . port : <EOL> try : <EOL> self . socket . bind ( ( self . addr , self . port ) ) <EOL> except socket . error as e : <EOL> print "<STR_LIT>" % self . port <EOL> print "<STR_LIT>" , e <EOL> self . _graceexit ( <NUM_LIT:1> ) <EOL> else : <EOL> for i in range ( <NUM_LIT:50> ) : <EOL> self . port = random . randint ( <NUM_LIT> , <NUM_LIT> ) <EOL> try : <EOL> self . socket . bind ( ( self . addr , self . port ) ) <EOL> break <EOL> except socket . error as e : <EOL> print "<STR_LIT>" , e <EOL> self . socket . listen ( <NUM_LIT:10> ) <EOL> self . connectionpool = ConnectionPool ( ) <EOL> try : <EOL> self . connectionpool . epoll = select . epoll ( ) <EOL> except AttributeError as e : <EOL> self . connectionpool . epoll = None <EOL> if logger : <EOL> LOGGERNODE = logger <EOL> else : <EOL> LOGGERNODE = None <EOL> self . replicas = { } <EOL> if self . isnameserver : <EOL> self . type = NODE_NAMESERVER <EOL> try : <EOL> self . nameserver = Nameserver ( self . addr , self . domain , self . useroute53 , <EOL> self . replicas , self . debug ) <EOL> except Exception as e : <EOL> print "<STR_LIT>" , e <EOL> print "<STR_LIT>" <EOL> self . _graceexit ( <NUM_LIT:1> ) <EOL> else : <EOL> self . type = NODE_REPLICA <EOL> self . alive = True <EOL> self . me = Peer ( self . addr , self . port , self . type ) <EOL> self . id = '<STR_LIT>' % ( self . addr , self . port ) <EOL> self . connectionpool . add_connection_to_self ( self . me , <EOL> SelfConnection ( self . receivedmessages , <EOL> self . receivedmessages_semaphore ) ) <EOL> self . logger = Logger ( "<STR_LIT>" % ( node_names [ self . type ] , self . id ) , lognode = LOGGERNODE ) <EOL> if self . isnameserver : <EOL> self . nameserver . add_logger ( self . logger ) <EOL> print "<STR_LIT>" % ( node_names [ self . type ] , self . id ) <EOL> self . nodeliveness = { } <EOL> self . bootstrapset = set ( ) <EOL> if givenbootstraplist : <EOL> self . discoverbootstrap ( givenbootstraplist ) <EOL> self . connecttobootstrap ( ) <EOL> self . stateuptodate = False <EOL> def _getipportpairs ( self , bootaddr , bootport ) : <EOL> for node in socket . getaddrinfo ( bootaddr , bootport , socket . AF_INET , socket . SOCK_STREAM ) : <EOL> yield Peer ( node [ <NUM_LIT:4> ] [ <NUM_LIT:0> ] , bootport , NODE_REPLICA ) <EOL> def discoverbootstrap ( self , givenbootstraplist ) : <EOL> bootstrapstrlist = givenbootstraplist . split ( "<STR_LIT:U+002C>" ) <EOL> for bootstrap in bootstrapstrlist : <EOL> if bootstrap . find ( "<STR_LIT::>" ) >= <NUM_LIT:0> : <EOL> bootaddr , bootport = bootstrap . split ( "<STR_LIT::>" ) <EOL> for peer in self . _getipportpairs ( bootaddr , int ( bootport ) ) : <EOL> self . bootstrapset . add ( peer ) <EOL> else : <EOL> answers = [ ] <EOL> try : <EOL> answers = dns . resolver . query ( '<STR_LIT>' + bootstrap , '<STR_LIT>' ) <EOL> except ( dns . resolver . NXDOMAIN , dns . exception . Timeout ) : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" % str ( bootstrap ) ) <EOL> for rdata in answers : <EOL> for peer in self . _getipportpairs ( str ( rdata . target ) , rdata . port ) : <EOL> self . bootstrapset . append ( peer ) <EOL> def connecttobootstrap ( self ) : <EOL> tries = <NUM_LIT:0> <EOL> keeptrying = True <EOL> while tries < BOOTSTRAPCONNECTTIMEOUT and keeptrying : <EOL> for bootpeer in self . bootstrapset : <EOL> try : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" % str ( bootpeer ) ) <EOL> helomessage = create_message ( MSG_HELO , self . me ) <EOL> successid = self . send ( helomessage , peer = bootpeer ) <EOL> if successid < <NUM_LIT:0> : <EOL> tries += <NUM_LIT:1> <EOL> continue <EOL> keeptrying = False <EOL> break <EOL> except socket . error as e : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" % str ( e ) ) <EOL> tries += <NUM_LIT:1> <EOL> continue <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> def startservice ( self ) : <EOL> receiver_thread = Thread ( target = self . server_loop , name = '<STR_LIT>' ) <EOL> receiver_thread . start ( ) <EOL> main_thread = Thread ( target = self . handle_messages , name = '<STR_LIT>' ) <EOL> main_thread . start ( ) <EOL> ping_thread = Timer ( LIVENESSTIMEOUT , self . ping_neighbor ) <EOL> ping_thread . name = '<STR_LIT>' <EOL> ping_thread . start ( ) <EOL> nascent_thread = Timer ( NASCENTTIMEOUT , self . clean_nascent ) <EOL> nascent_thread . name = '<STR_LIT>' <EOL> nascent_thread . start ( ) <EOL> if self . debug : <EOL> input_thread = Thread ( target = self . get_user_input_from_shell , name = '<STR_LIT>' ) <EOL> input_thread . start ( ) <EOL> return self <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( node_names [ self . type ] , self . addr , self . port ) <EOL> def statestr ( self ) : <EOL> returnstr = "<STR_LIT>" <EOL> for peer in self . replicas : <EOL> returnstr += node_names [ peer . type ] + "<STR_LIT>" % ( peer . addr , peer . port ) <EOL> if hasattr ( self , '<STR_LIT>' ) and len ( self . pendingcommands ) > <NUM_LIT:0> : <EOL> pending = "<STR_LIT>" . join ( "<STR_LIT>" % ( cno , proposal ) for cno , proposal in self . pendingcommands . iteritems ( ) ) <EOL> returnstr = "<STR_LIT>" % ( returnstr , pending ) <EOL> return returnstr <EOL> def ping_neighbor ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> for peer in self . replicas : <EOL> if peer == self . me : <EOL> continue <EOL> if peer in self . nodeliveness : <EOL> nosound = time . time ( ) - self . nodeliveness [ peer ] <EOL> else : <EOL> nosound = LIVENESSTIMEOUT + <NUM_LIT:1> <EOL> if nosound <= LIVENESSTIMEOUT : <EOL> self . replicas [ peer ] = <NUM_LIT:0> <EOL> continue <EOL> if nosound > LIVENESSTIMEOUT : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" % str ( peer ) ) <EOL> pingmessage = create_message ( MSG_PING , self . me ) <EOL> successid = self . send ( pingmessage , peer = peer ) <EOL> if successid < <NUM_LIT:0> or nosound > ( <NUM_LIT:2> * LIVENESSTIMEOUT ) : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) <EOL> self . replicas [ peer ] += <NUM_LIT:1> <EOL> time . sleep ( LIVENESSTIMEOUT ) <EOL> def clean_nascent ( self ) : <EOL> lastnascentset = set ( [ ] ) <EOL> while True : <EOL> for sock in lastnascentset . intersection ( self . connectionpool . nascentsockets ) : <EOL> self . connectionpool . activesockets . remove ( sock ) <EOL> self . connectionpool . nascentsockets . remove ( sock ) <EOL> lastnascentset = self . connectionpool . nascentsockets <EOL> time . sleep ( NASCENTTIMEOUT ) <EOL> def server_loop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . socket . listen ( <NUM_LIT:10> ) <EOL> if self . connectionpool . epoll : <EOL> self . connectionpool . epoll . register ( self . socket . fileno ( ) , select . EPOLLIN ) <EOL> self . use_epoll ( ) <EOL> else : <EOL> self . connectionpool . activesockets . add ( self . socket ) <EOL> self . use_select ( ) <EOL> self . socket . close ( ) <EOL> return <EOL> def use_epoll ( self ) : <EOL> while self . alive : <EOL> try : <EOL> events = self . connectionpool . epoll . poll ( <NUM_LIT:1> ) <EOL> for fileno , event in events : <EOL> if fileno == self . socket . fileno ( ) : <EOL> clientsock , clientaddr = self . socket . accept ( ) <EOL> clientsock . setblocking ( <NUM_LIT:0> ) <EOL> self . connectionpool . epoll . register ( clientsock . fileno ( ) , select . EPOLLIN ) <EOL> self . connectionpool . epollsockets [ clientsock . fileno ( ) ] = clientsock <EOL> elif event & select . EPOLLIN : <EOL> success = self . handle_connection ( self . connectionpool . epollsockets [ fileno ] ) <EOL> if not success : <EOL> self . connectionpool . epoll . unregister ( fileno ) <EOL> self . connectionpool . del_connection_by_socket ( self . connectionpool . epollsockets [ fileno ] ) <EOL> self . connectionpool . epollsockets [ fileno ] . close ( ) <EOL> del self . connectionpool . epollsockets [ fileno ] <EOL> elif event & select . EPOLLHUP : <EOL> self . connectionpool . epoll . unregister ( fileno ) <EOL> self . connectionpool . epollsockets [ fileno ] . close ( ) <EOL> del self . connectionpool . epollsockets [ fileno ] <EOL> except KeyboardInterrupt , EOFError : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> self . connectionpool . epoll . unregister ( self . socket . fileno ( ) ) <EOL> self . connectionpool . epoll . close ( ) <EOL> def use_select ( self ) : <EOL> while self . alive : <EOL> try : <EOL> inputready , outputready , exceptready = select . select ( self . connectionpool . activesockets , <EOL> [ ] , <EOL> self . connectionpool . activesockets , <EOL> <NUM_LIT:1> ) <EOL> for s in exceptready : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT:%s>" % s ) <EOL> for s in inputready : <EOL> if s == self . socket : <EOL> clientsock , clientaddr = self . socket . accept ( ) <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> % str ( clientaddr ) ) <EOL> self . connectionpool . activesockets . add ( clientsock ) <EOL> self . connectionpool . nascentsockets . add ( clientsock ) <EOL> success = True <EOL> else : <EOL> success = self . handle_connection ( s ) <EOL> if not success : <EOL> self . connectionpool . del_connection_by_socket ( s ) <EOL> except KeyboardInterrupt , EOFError : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def handle_connection ( self , clientsock ) : <EOL> """<STR_LIT>""" <EOL> connection = self . connectionpool . get_connection_by_socket ( clientsock ) <EOL> try : <EOL> for message in connection . received_bytes ( ) : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" % str ( message ) ) <EOL> if message . type == MSG_STATUS : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" <EOL> % self . __str__ ( ) ) <EOL> messagestr = pickle . dumps ( self . __str__ ( ) ) <EOL> message = struct . pack ( "<STR_LIT:I>" , len ( messagestr ) ) + messagestr <EOL> clientsock . send ( message ) <EOL> return False <EOL> if message . source . type != NODE_CLIENT : <EOL> self . nodeliveness [ message . source ] = time . time ( ) <EOL> if message . source in self . replicas : <EOL> self . replicas [ message . source ] = <NUM_LIT:0> <EOL> self . receivedmessages . append ( ( message , connection ) ) <EOL> self . receivedmessages_semaphore . release ( ) <EOL> if message . type == MSG_CLIENTREQUEST or message . type == MSG_INCCLIENTREQUEST : <EOL> self . connectionpool . add_connection_to_peer ( message . source , connection ) <EOL> elif message . type in ( MSG_HELO , MSG_HELOREPLY , MSG_UPDATE ) : <EOL> self . connectionpool . add_connection_to_peer ( message . source , connection ) <EOL> return True <EOL> except ConnectionError as e : <EOL> return False <EOL> def handle_messages ( self ) : <EOL> while True : <EOL> self . receivedmessages_semaphore . acquire ( ) <EOL> ( message_to_process , connection ) = self . receivedmessages . pop ( <NUM_LIT:0> ) <EOL> if message_to_process . type == MSG_CLIENTREQUEST : <EOL> if message_to_process . clientbatch : <EOL> self . process_message ( message_to_process , connection ) <EOL> continue <EOL> msgconns = [ ( message_to_process , connection ) ] <EOL> for m , c in self . receivedmessages : <EOL> if m . type == MSG_CLIENTREQUEST : <EOL> self . receivedmessages_semaphore . acquire ( ) <EOL> self . receivedmessages . remove ( ( m , c ) ) <EOL> msgconns . append ( ( m , c ) ) <EOL> if len ( msgconns ) > <NUM_LIT:1> : <EOL> self . process_messagelist ( msgconns ) <EOL> else : <EOL> self . process_message ( message_to_process , connection ) <EOL> else : <EOL> self . process_message ( message_to_process , connection ) <EOL> return <EOL> def process_messagelist ( self , msgconnlist ) : <EOL> """<STR_LIT>""" <EOL> with self . lock : <EOL> self . msg_clientrequest_batch ( msgconnlist ) <EOL> return True <EOL> def process_message ( self , message , connection ) : <EOL> """<STR_LIT>""" <EOL> mname = "<STR_LIT>" % msg_names [ message . type ] . lower ( ) <EOL> try : <EOL> method = getattr ( self , mname ) <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" % mname ) <EOL> except AttributeError : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" % mname ) <EOL> return False <EOL> with self . lock : <EOL> method ( connection , message ) <EOL> return True <EOL> def msg_helo ( self , conn , msg ) : <EOL> return <EOL> def msg_heloreply ( self , conn , msg ) : <EOL> if msg . leader : <EOL> if msg . leader == self . me : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> return <EOL> else : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> if msg . source in self . bootstrapset : <EOL> self . bootstrapset . remove ( msg . source ) <EOL> self . bootstrapset . add ( msg . leader ) <EOL> self . connecttobootstrap ( ) <EOL> def msg_ping ( self , conn , msg ) : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> pingreplymessage = create_message ( MSG_PINGREPLY , self . me ) <EOL> conn . send ( pingreplymessage ) <EOL> def msg_pingreply ( self , conn , msg ) : <EOL> return <EOL> def cmd_help ( self , args ) : <EOL> """<STR_LIT>""" <EOL> print "<STR_LIT>" <EOL> for attr in dir ( self ) : <EOL> if attr . startswith ( "<STR_LIT>" ) : <EOL> print attr . replace ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def cmd_exit ( self , args ) : <EOL> """<STR_LIT>""" <EOL> self . alive = False <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def cmd_state ( self , args ) : <EOL> """<STR_LIT>""" <EOL> print "<STR_LIT>" % ( self . statestr ( ) ) <EOL> def get_user_input_from_shell ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . alive : <EOL> try : <EOL> input = raw_input ( "<STR_LIT:>>" ) <EOL> if len ( input ) == <NUM_LIT:0> : <EOL> continue <EOL> else : <EOL> input = input . split ( ) <EOL> mname = "<STR_LIT>" % input [ <NUM_LIT:0> ] . lower ( ) <EOL> try : <EOL> method = getattr ( self , mname ) <EOL> except AttributeError as e : <EOL> print "<STR_LIT>" , str ( e ) <EOL> continue <EOL> with self . lock : <EOL> method ( input ) <EOL> except KeyboardInterrupt : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> except EOFError : <EOL> return <EOL> return <EOL> def send ( self , message , peer = None , group = None ) : <EOL> if peer : <EOL> connection = self . connectionpool . get_connection_by_peer ( peer ) <EOL> if connection == None : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" % str ( peer ) ) <EOL> return - <NUM_LIT:1> <EOL> connection . send ( message ) <EOL> return message [ FLD_ID ] <EOL> elif group : <EOL> ids = [ ] <EOL> for peer , liveness in group . iteritems ( ) : <EOL> connection = self . connectionpool . get_connection_by_peer ( peer ) <EOL> if connection == None : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , <EOL> "<STR_LIT>" % str ( peer ) ) <EOL> continue <EOL> connection . send ( message ) <EOL> ids . append ( message [ FLD_ID ] ) <EOL> message [ FLD_ID ] = assignuniqueid ( ) <EOL> return ids <EOL> def terminate_handler ( self , signal , frame ) : <EOL> if self . debug : self . logger . write ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> self . logger . close ( ) <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . flush ( ) <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def _graceexit ( self , exitcode = <NUM_LIT:0> ) : <EOL> sys . stdout . flush ( ) <EOL> sys . stderr . flush ( ) <EOL> if hasattr ( self , '<STR_LIT>' ) : self . logger . close ( ) <EOL> os . _exit ( exitcode ) </s>
