<s> """<STR_LIT>""" <EOL> import exceptions <EOL> import random as _random <EOL> import time <EOL> import sys <EOL> import os <EOL> import socket <EOL> from gogreen import coro <EOL> import pyinfo <EOL> import error <EOL> class ServerUnavailable ( Exception ) : <EOL> pass <EOL> class ServerError ( ServerUnavailable ) : <EOL> pass <EOL> class ClientError ( ServerUnavailable ) : <EOL> pass <EOL> class SplitAccessError ( Exception ) : <EOL> pass <EOL> class SplitServerError ( ServerError , SplitAccessError ) : <EOL> pass <EOL> class SplitClientError ( ClientError , SplitAccessError ) : <EOL> pass <EOL> class LookAside ( Exception ) : <EOL> pass <EOL> BLANK_TOKEN = None <EOL> CORO_LOCAL_TDATA = '<STR_LIT>' <EOL> CORO_LOCAL_TCTRL = '<STR_LIT>' <EOL> CORO_LOCAL_SOURCE = '<STR_LIT>' <EOL> DEFAULT_RETRY = <NUM_LIT:2> <EOL> ACCESS_TRACE_LIMITS = { '<STR_LIT>' : None , '<STR_LIT>' : None , '<STR_LIT>' : None } <EOL> ACCESS_TRACE_OFF = <NUM_LIT:0> <EOL> ACCESS_TRACE_TERSE = <NUM_LIT:10> <EOL> ACCESS_TRACE_INFO = <NUM_LIT:20> <EOL> ACCESS_TRACE_DEBUG = <NUM_LIT:30> <EOL> ACCESS_TRACE_VERBOSE = <NUM_LIT> <EOL> def _tlb_status ( object ) : <EOL> return coro . get_local ( '<STR_LIT>' % object , False ) <EOL> def _set_trace_local ( value , clear ) : <EOL> if not coro . has_local ( CORO_LOCAL_TCTRL ) : <EOL> coro . set_local ( CORO_LOCAL_TCTRL , { } ) <EOL> data = coro . get_local ( CORO_LOCAL_TCTRL ) <EOL> data [ '<STR_LIT:value>' ] = value <EOL> data [ '<STR_LIT>' ] = clear <EOL> def _get_trace_local ( ) : <EOL> data = coro . get_local ( CORO_LOCAL_TCTRL , { } ) <EOL> return data . get ( '<STR_LIT:value>' , False ) <EOL> def _clr_trace_local ( ) : <EOL> if coro . get_local ( CORO_LOCAL_TCTRL , { } ) . get ( '<STR_LIT>' , False ) : <EOL> data = coro . pop_local ( CORO_LOCAL_TCTRL , { } ) <EOL> else : <EOL> data = coro . get_local ( CORO_LOCAL_TCTRL , { } ) <EOL> return data . get ( '<STR_LIT:value>' , False ) <EOL> def slave_read ( status = None ) : <EOL> if status is None : <EOL> return coro . get_local ( '<STR_LIT>' , False ) <EOL> else : <EOL> return coro . set_local ( '<STR_LIT>' , status ) <EOL> def trace_dump ( clear = True ) : <EOL> '''<STR_LIT>''' <EOL> if clear : <EOL> tdata = coro . pop_local ( CORO_LOCAL_TDATA , { } ) <EOL> tlevl = _clr_trace_local ( ) <EOL> else : <EOL> tdata = coro . get_local ( CORO_LOCAL_TDATA , { } ) <EOL> tlevl = _get_trace_local ( ) <EOL> if not tlevl : <EOL> return None <EOL> total , idcnt , count = <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> <EOL> for obj , data in tdata . items ( ) : <EOL> for cmd , ( elapse , ids , cnt ) in data . items ( ) : <EOL> total += elapse <EOL> count += cnt <EOL> idcnt += ids <EOL> if tlevl > ACCESS_TRACE_TERSE : <EOL> print '<STR_LIT>' % ( <EOL> elapse , cnt , ids , obj , cmd ) <EOL> if not total : <EOL> return None <EOL> lmt = has_trace_limits ( ) <EOL> if lmt is None : <EOL> lmt = '<STR_LIT>' <EOL> else : <EOL> lmt = '<STR_LIT>' % lmt <EOL> print '<STR_LIT>' % ( <EOL> total , count , idcnt , lmt ) <EOL> def enable_trace ( value , clear = True ) : <EOL> '''<STR_LIT>''' <EOL> _set_trace_local ( value , clear ) <EOL> return _get_trace_local ( ) , value <EOL> sum_tuple = lambda * a : tuple ( map ( sum , zip ( * a ) ) ) <EOL> def _trace_data ( start , obj , cmd , vids , ** kwargs ) : <EOL> if not _get_trace_local ( ) : <EOL> return None <EOL> if not _trace_check_limits ( obj , cmd , vids ) : <EOL> return None <EOL> elapse = time . time ( ) - start <EOL> if not coro . has_local ( CORO_LOCAL_TDATA ) : <EOL> coro . set_local ( CORO_LOCAL_TDATA , { } ) <EOL> data = coro . get_local ( CORO_LOCAL_TDATA ) <EOL> data = data . setdefault ( obj , { } ) <EOL> look = not isinstance ( vids , ( list , tuple ) ) and <NUM_LIT:1> or len ( vids ) <EOL> data [ cmd ] = sum_tuple ( data . get ( cmd , ( <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> ) ) , ( elapse , look , <NUM_LIT:1> ) ) <EOL> if _get_trace_local ( ) > ACCESS_TRACE_DEBUG : <EOL> stack = pyinfo . rawstack ( ) <EOL> stack = stack [ : - <NUM_LIT:5> ] <EOL> while stack and stack [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) : <EOL> stack . pop ( <NUM_LIT:0> ) <EOL> stack = '<STR_LIT>' % '<STR_LIT:|>' . join ( [ '<STR_LIT>' % x for x in stack ] ) <EOL> else : <EOL> stack = '<STR_LIT>' <EOL> if _get_trace_local ( ) < ACCESS_TRACE_DEBUG : <EOL> return None <EOL> print '<STR_LIT>' % ( <EOL> elapse , obj , cmd , vids , <EOL> kwargs . get ( '<STR_LIT:args>' , ( ) ) , kwargs . get ( '<STR_LIT>' , { } ) , <EOL> stack ) <EOL> def _execute_trace_decorator ( ) : <EOL> '''<STR_LIT>''' <EOL> def function ( method ) : <EOL> def tracer ( obj , n , cmd , vids , * args , ** kwargs ) : <EOL> start = _get_trace_local ( ) and time . time ( ) or <NUM_LIT:0> <EOL> try : <EOL> return method ( obj , n , cmd , vids , * args , ** kwargs ) <EOL> finally : <EOL> try : <EOL> _trace_data ( start , obj , cmd , vids , ** kwargs ) <EOL> except : <EOL> raise <EOL> return tracer <EOL> return function <EOL> def _complete_trace_decorator ( ) : <EOL> '''<STR_LIT>''' <EOL> def function ( method ) : <EOL> def tracer ( * args , ** kwargs ) : <EOL> start = _get_trace_local ( ) and time . time ( ) or <NUM_LIT:0> <EOL> try : <EOL> return method ( * args , ** kwargs ) <EOL> finally : <EOL> try : <EOL> _trace_data ( <EOL> start , <EOL> args [ <NUM_LIT:0> ] . __class__ . __name__ , <EOL> method . __name__ , <EOL> [ ] ) <EOL> except : <EOL> raise <EOL> return tracer <EOL> return function <EOL> _trace_limits = ACCESS_TRACE_LIMITS <EOL> def trace_limit ( obj = None , cmd = None , vids = None ) : <EOL> '''<STR_LIT>''' <EOL> _trace_limits [ '<STR_LIT>' ] = obj <EOL> _trace_limits [ '<STR_LIT>' ] = cmd <EOL> _trace_limits [ '<STR_LIT>' ] = vids <EOL> def trace_limits_clear ( ) : <EOL> global _trace_limits <EOL> _trace_limits = { '<STR_LIT>' : None , '<STR_LIT>' : None , '<STR_LIT>' : None } <EOL> def has_trace_limits ( ) : <EOL> '''<STR_LIT>''' <EOL> r = '<STR_LIT>' . join ( [ str ( x [ <NUM_LIT:0> ] ) [ <NUM_LIT:0> ] for x in _trace_limits . items ( ) if x [ <NUM_LIT:1> ] is not None ] ) or None <EOL> return r <EOL> def _trace_check_limits ( obj , cmd , vids ) : <EOL> obj_limit = _trace_limits . get ( '<STR_LIT>' ) <EOL> if obj_limit is not None and obj_limit != obj : <EOL> return False <EOL> cmd_limit = _trace_limits . get ( '<STR_LIT>' ) <EOL> if cmd_limit is not None and cmd_limit != cmd : <EOL> return False <EOL> vids_l = _trace_limits . get ( '<STR_LIT>' ) <EOL> vids_l = ( isinstance ( vids_l , ( list , tuple ) ) and [ vids_l ] or [ [ vids_l ] ] ) [ <NUM_LIT:0> ] <EOL> vids_l = set ( filter ( lambda x : x is not None , vids_l ) ) <EOL> vids = ( isinstance ( vids , ( list , tuple ) ) and [ vids ] or [ [ vids ] ] ) [ <NUM_LIT:0> ] <EOL> vids = set ( filter ( lambda x : x is not None , vids ) ) <EOL> if vids_l and not vids_l . intersection ( vids ) : <EOL> return False <EOL> return True <EOL> def _unwrap ( result ) : <EOL> '''<STR_LIT>''' <EOL> if not isinstance ( result , dict ) : <EOL> return result <EOL> if not result . get ( '<STR_LIT>' , False ) : <EOL> return result <EOL> if not result . get ( '<STR_LIT>' , <NUM_LIT:0> ) : <EOL> return result . get ( '<STR_LIT:result>' , result ) <EOL> raise error . lookup ( result . get ( '<STR_LIT>' ) , * result . get ( '<STR_LIT:args>' , ( ) ) ) <EOL> @ _execute_trace_decorator ( ) <EOL> def execute ( <EOL> object , notifier , command , vids , <EOL> args = ( ) , kwargs = { } , timeout = None , retry = None , raw = False ) : <EOL> if not vids and isinstance ( vids , list ) : <EOL> return [ ] <EOL> if retry is None : <EOL> retry = DEFAULT_RETRY <EOL> tlb = '<STR_LIT>' % object <EOL> val = coro . get_local ( tlb ) <EOL> if val is None : <EOL> val = False <EOL> elif not val : <EOL> val = True <EOL> else : <EOL> raise LookAside ( ( tlb , val ) ) <EOL> look = ( isinstance ( vids , ( list , tuple ) ) and [ vids ] or [ [ vids ] ] ) [ <NUM_LIT:0> ] <EOL> data = { <EOL> '<STR_LIT>' : ( tlb , val ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , slave_read ( ) ) , <EOL> '<STR_LIT:source>' : socket . gethostname ( ) , <EOL> '<STR_LIT>' : command , <EOL> '<STR_LIT:args>' : args , <EOL> '<STR_LIT>' : kwargs } <EOL> retry += <NUM_LIT:1> <EOL> results = map ( lambda i : None , xrange ( len ( look ) ) ) <EOL> pending = { } <EOL> if not look : <EOL> return results <EOL> filter ( <EOL> lambda i : pending . setdefault ( i [ <NUM_LIT:0> ] , set ( [ ] ) ) . add ( i [ <NUM_LIT:1> ] ) , <EOL> zip ( look , xrange ( len ( look ) ) ) ) <EOL> for pos in pending . pop ( BLANK_TOKEN , [ ] ) : <EOL> results [ pos ] = [ { } ] <EOL> while retry and pending : <EOL> retry -= <NUM_LIT:1> <EOL> lookup = pending . keys ( ) <EOL> values = notifier . rpcs ( <EOL> object , lookup , '<STR_LIT>' , data , <EOL> timeout = timeout , <EOL> local = not val ) <EOL> if not values : <EOL> continue <EOL> for i in xrange ( len ( lookup ) ) : <EOL> if not values [ i ] : <EOL> push = pending . get ( lookup [ i ] ) <EOL> else : <EOL> push = pending . pop ( lookup [ i ] ) <EOL> for pos in push : <EOL> results [ pos ] = values [ i ] <EOL> errors = filter ( lambda i : not i , results ) <EOL> if errors : <EOL> if errors [ <NUM_LIT:0> ] is None : <EOL> raise ClientError ( look , data , errors ) <EOL> else : <EOL> raise ServerError ( look , data , errors ) <EOL> if not raw : <EOL> results = map ( lambda i : _unwrap ( ( i and [ i [ <NUM_LIT:0> ] ] or [ None ] ) [ <NUM_LIT:0> ] ) , results ) <EOL> else : <EOL> results = map ( lambda i : map ( _unwrap , i ) , results ) <EOL> if not isinstance ( vids , list ) : <EOL> results = results [ <NUM_LIT:0> ] <EOL> return results <EOL> def random ( <EOL> object , notifier , command , args = ( ) , kwargs = { } , <EOL> timeout = None , retry = None , raw = False ) : <EOL> return execute ( <EOL> object , notifier , command , <EOL> _random . randint ( <NUM_LIT:0> , ( <NUM_LIT:8> << <NUM_LIT> ) - <NUM_LIT:1> ) , <EOL> args , kwargs , timeout , retry , raw ) <EOL> def _flatten ( data ) : <EOL> if isinstance ( data , ( int , long ) ) : <EOL> return ( data , ) <EOL> if not data : <EOL> return ( ) <EOL> result = reduce ( lambda x , y : _flatten ( x ) + _flatten ( y ) , data ) <EOL> if isinstance ( result , ( tuple , list ) ) : <EOL> return tuple ( result ) <EOL> else : <EOL> return ( result , ) <EOL> def _fatten ( request , result ) : <EOL> if isinstance ( request , ( int , long ) ) : <EOL> return result [ request ] <EOL> else : <EOL> return map ( lambda i : _fatten ( i , result ) , request ) <EOL> class SplitAccess ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , notifier , ** kwargs ) : <EOL> self . _notifier = notifier <EOL> self . _retry = kwargs . get ( '<STR_LIT>' , DEFAULT_RETRY ) <EOL> def __repr__ ( self ) : <EOL> return repr ( self . _notifier ) <EOL> def publish ( self , object , id , cmd , args ) : <EOL> return self . _notifier . publish ( object , id , cmd , args ) <EOL> def rpcs ( self , obj , vid_list , cmd , args , * more , ** kwargs ) : <EOL> if more : kwargs [ '<STR_LIT>' ] = more [ <NUM_LIT:0> ] <EOL> seq , cnt = self . _notifier . rpcs_push ( obj , vid_list , cmd , args , ** kwargs ) <EOL> return map ( lambda i : [ i ] , range ( seq , seq + cnt ) ) <EOL> def rpc ( self , obj , vid , cmd , args , * more , ** kwargs ) : <EOL> result = self . rpcs ( obj , [ vid ] , cmd , args , * more , ** kwargs ) <EOL> if result : <EOL> return result [ <NUM_LIT:0> ] <EOL> else : <EOL> return [ <NUM_LIT:0> ] <EOL> @ _complete_trace_decorator ( ) <EOL> def complete ( self , seq_list , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> if not seq_list : <EOL> return None <EOL> retry = self . _retry + <NUM_LIT:1> <EOL> results = map ( lambda i : None , xrange ( len ( seq_list ) ) ) <EOL> pending = dict ( zip ( seq_list , xrange ( len ( seq_list ) ) ) ) <EOL> errored = { } <EOL> if len ( results ) != len ( pending ) : <EOL> raise ValueError ( <EOL> '<STR_LIT>' , <EOL> len ( results ) , <EOL> len ( pending ) ) <EOL> while retry and pending : <EOL> retry -= <NUM_LIT:1> <EOL> result = self . _notifier . rpcs_recv ( pending . keys ( ) , ** kwargs ) <EOL> for seq , req , rep in result : <EOL> pos = pending . pop ( seq ) <EOL> tmp = errored . pop ( pos , None ) <EOL> results [ pos ] = rep <EOL> if rep or not req : <EOL> continue <EOL> seq = self . rpc ( * req , ** kwargs ) . pop ( ) <EOL> if not seq : <EOL> continue <EOL> pending [ seq ] = pos <EOL> errored [ pos ] = ( req , rep ) <EOL> if not errored : <EOL> return map ( lambda i : _unwrap ( i [ <NUM_LIT:0> ] ) , results ) <EOL> errored = dict ( map ( <EOL> lambda i : ( seq_list [ i [ <NUM_LIT:0> ] ] , i [ <NUM_LIT:1> ] ) , <EOL> errored . iteritems ( ) ) ) <EOL> if filter ( lambda i : i [ <NUM_LIT:1> ] is None , errored . itervalues ( ) ) : <EOL> raise SplitClientError ( seq_list , results , errored ) <EOL> raise SplitServerError ( seq_list , results , errored ) <EOL> def completes ( self , * args , ** kwargs ) : <EOL> '''<STR_LIT>''' <EOL> if not args : <EOL> return None <EOL> flat = _flatten ( args ) <EOL> result = _fatten ( args , dict ( zip ( flat , self . complete ( flat , ** kwargs ) ) ) ) <EOL> if len ( args ) == <NUM_LIT:1> : <EOL> return result . pop ( ) <EOL> else : <EOL> return result <EOL> @ _complete_trace_decorator ( ) <EOL> def any ( self , seq_set ) : <EOL> '''<STR_LIT>''' <EOL> if not seq_set : <EOL> raise ValueError ( '<STR_LIT>' , seq_set ) <EOL> seq , req , rep = self . _notifier . rpcs_pop ( seq_set ) <EOL> if rep is None : <EOL> raise ClientError ( seq , req , rep ) <EOL> if not rep : <EOL> raise ServerError ( seq , req , rep ) <EOL> seq_set . remove ( seq ) <EOL> return ( seq , _unwrap ( rep . pop ( ) ) ) <EOL> def clear ( self , seqs ) : <EOL> '''<STR_LIT>''' <EOL> return self . _notifier . rpcs_clear ( seqs ) </s>
