<s> """<STR_LIT>""" <EOL> import threading <EOL> import socket <EOL> import Queue <EOL> import select <EOL> import struct <EOL> import logging <EOL> from thrift . transport import TTransport <EOL> from thrift . protocol . TBinaryProtocol import TBinaryProtocolFactory <EOL> __all__ = [ '<STR_LIT>' ] <EOL> class Worker ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , queue ) : <EOL> threading . Thread . __init__ ( self ) <EOL> self . queue = queue <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> while True : <EOL> try : <EOL> processor , iprot , oprot , otrans , callback = self . queue . get ( ) <EOL> if processor is None : <EOL> break <EOL> processor . process ( iprot , oprot ) <EOL> callback ( True , otrans . getvalue ( ) ) <EOL> except Exception : <EOL> logging . exception ( "<STR_LIT>" ) <EOL> callback ( False , '<STR_LIT>' ) <EOL> WAIT_LEN = <NUM_LIT:0> <EOL> WAIT_MESSAGE = <NUM_LIT:1> <EOL> WAIT_PROCESS = <NUM_LIT:2> <EOL> SEND_ANSWER = <NUM_LIT:3> <EOL> CLOSED = <NUM_LIT:4> <EOL> def locked ( func ) : <EOL> """<STR_LIT>""" <EOL> def nested ( self , * args , ** kwargs ) : <EOL> self . lock . acquire ( ) <EOL> try : <EOL> return func ( self , * args , ** kwargs ) <EOL> finally : <EOL> self . lock . release ( ) <EOL> return nested <EOL> def socket_exception ( func ) : <EOL> """<STR_LIT>""" <EOL> def read ( self , * args , ** kwargs ) : <EOL> try : <EOL> return func ( self , * args , ** kwargs ) <EOL> except socket . error : <EOL> self . close ( ) <EOL> return read <EOL> class Connection : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , new_socket , wake_up ) : <EOL> self . socket = new_socket <EOL> self . socket . setblocking ( False ) <EOL> self . status = WAIT_LEN <EOL> self . len = <NUM_LIT:0> <EOL> self . message = '<STR_LIT>' <EOL> self . lock = threading . Lock ( ) <EOL> self . wake_up = wake_up <EOL> def _read_len ( self ) : <EOL> """<STR_LIT>""" <EOL> read = self . socket . recv ( <NUM_LIT:4> - len ( self . message ) ) <EOL> if len ( read ) == <NUM_LIT:0> : <EOL> if len ( self . message ) != <NUM_LIT:0> : <EOL> logging . error ( "<STR_LIT>" ) <EOL> self . close ( ) <EOL> return <EOL> self . message += read <EOL> if len ( self . message ) == <NUM_LIT:4> : <EOL> self . len , = struct . unpack ( '<STR_LIT>' , self . message ) <EOL> if self . len < <NUM_LIT:0> : <EOL> logging . error ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . close ( ) <EOL> elif self . len == <NUM_LIT:0> : <EOL> logging . error ( "<STR_LIT>" ) <EOL> self . close ( ) <EOL> else : <EOL> self . message = '<STR_LIT>' <EOL> self . status = WAIT_MESSAGE <EOL> @ socket_exception <EOL> def read ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . status in ( WAIT_LEN , WAIT_MESSAGE ) <EOL> if self . status == WAIT_LEN : <EOL> self . _read_len ( ) <EOL> elif self . status == WAIT_MESSAGE : <EOL> read = self . socket . recv ( self . len - len ( self . message ) ) <EOL> if len ( read ) == <NUM_LIT:0> : <EOL> logging . error ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( len ( self . message ) , self . len ) ) <EOL> self . close ( ) <EOL> return <EOL> self . message += read <EOL> if len ( self . message ) == self . len : <EOL> self . status = WAIT_PROCESS <EOL> @ socket_exception <EOL> def write ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . status == SEND_ANSWER <EOL> sent = self . socket . send ( self . message ) <EOL> if sent == len ( self . message ) : <EOL> self . status = WAIT_LEN <EOL> self . message = '<STR_LIT>' <EOL> self . len = <NUM_LIT:0> <EOL> else : <EOL> self . message = self . message [ sent : ] <EOL> @ locked <EOL> def ready ( self , all_ok , message ) : <EOL> """<STR_LIT>""" <EOL> assert self . status == WAIT_PROCESS <EOL> if not all_ok : <EOL> self . close ( ) <EOL> self . wake_up ( ) <EOL> return <EOL> self . len = '<STR_LIT>' <EOL> if len ( message ) == <NUM_LIT:0> : <EOL> self . message = '<STR_LIT>' <EOL> self . status = WAIT_LEN <EOL> else : <EOL> self . message = struct . pack ( '<STR_LIT>' , len ( message ) ) + message <EOL> self . status = SEND_ANSWER <EOL> self . wake_up ( ) <EOL> @ locked <EOL> def is_writeable ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . status == SEND_ANSWER <EOL> @ locked <EOL> def is_readable ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . status in ( WAIT_LEN , WAIT_MESSAGE ) <EOL> @ locked <EOL> def is_closed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . status == CLOSED <EOL> def fileno ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . socket . fileno ( ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> self . status = CLOSED <EOL> self . socket . close ( ) <EOL> class TNonblockingServer : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> processor , <EOL> lsocket , <EOL> inputProtocolFactory = None , <EOL> outputProtocolFactory = None , <EOL> threads = <NUM_LIT:10> ) : <EOL> self . processor = processor <EOL> self . socket = lsocket <EOL> self . in_protocol = inputProtocolFactory or TBinaryProtocolFactory ( ) <EOL> self . out_protocol = outputProtocolFactory or self . in_protocol <EOL> self . threads = int ( threads ) <EOL> self . clients = { } <EOL> self . tasks = Queue . Queue ( ) <EOL> self . _read , self . _write = socket . socketpair ( ) <EOL> self . prepared = False <EOL> self . _stop = False <EOL> def setNumThreads ( self , num ) : <EOL> """<STR_LIT>""" <EOL> assert not self . prepared , "<STR_LIT>" <EOL> self . threads = num <EOL> def prepare ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . prepared : <EOL> return <EOL> self . socket . listen ( ) <EOL> for _ in xrange ( self . threads ) : <EOL> thread = Worker ( self . tasks ) <EOL> thread . setDaemon ( True ) <EOL> thread . start ( ) <EOL> self . prepared = True <EOL> def wake_up ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _write . send ( '<STR_LIT:1>' ) <EOL> def stop ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _stop = True <EOL> self . wake_up ( ) <EOL> def _select ( self ) : <EOL> """<STR_LIT>""" <EOL> readable = [ self . socket . handle . fileno ( ) , self . _read . fileno ( ) ] <EOL> writable = [ ] <EOL> for i , connection in self . clients . items ( ) : <EOL> if connection . is_readable ( ) : <EOL> readable . append ( connection . fileno ( ) ) <EOL> if connection . is_writeable ( ) : <EOL> writable . append ( connection . fileno ( ) ) <EOL> if connection . is_closed ( ) : <EOL> del self . clients [ i ] <EOL> return select . select ( readable , writable , readable ) <EOL> def handle ( self ) : <EOL> """<STR_LIT>""" <EOL> assert self . prepared , "<STR_LIT>" <EOL> rset , wset , xset = self . _select ( ) <EOL> for readable in rset : <EOL> if readable == self . _read . fileno ( ) : <EOL> self . _read . recv ( <NUM_LIT> ) <EOL> elif readable == self . socket . handle . fileno ( ) : <EOL> client = self . socket . accept ( ) . handle <EOL> self . clients [ client . fileno ( ) ] = Connection ( client , <EOL> self . wake_up ) <EOL> else : <EOL> connection = self . clients [ readable ] <EOL> connection . read ( ) <EOL> if connection . status == WAIT_PROCESS : <EOL> itransport = TTransport . TMemoryBuffer ( connection . message ) <EOL> otransport = TTransport . TMemoryBuffer ( ) <EOL> iprot = self . in_protocol . getProtocol ( itransport ) <EOL> oprot = self . out_protocol . getProtocol ( otransport ) <EOL> self . tasks . put ( [ self . processor , iprot , oprot , <EOL> otransport , connection . ready ] ) <EOL> for writeable in wset : <EOL> self . clients [ writeable ] . write ( ) <EOL> for oob in xset : <EOL> self . clients [ oob ] . close ( ) <EOL> del self . clients [ oob ] <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> for _ in xrange ( self . threads ) : <EOL> self . tasks . put ( [ None , None , None , None , None ] ) <EOL> self . socket . close ( ) <EOL> self . prepared = False <EOL> def serve ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _stop = False <EOL> self . prepare ( ) <EOL> while not self . _stop : <EOL> self . handle ( ) </s>
