<s> """<STR_LIT>""" <EOL> import collections <EOL> import logging <EOL> import os <EOL> import sys <EOL> import threading <EOL> import types <EOL> from google . appengine . api . apiproxy_stub_map import UserRPC <EOL> from google . appengine . api . apiproxy_rpc import RPC <EOL> from google . appengine . datastore import datastore_rpc <EOL> from ndb import eventloop , utils <EOL> logging_debug = utils . logging_debug <EOL> def is_generator ( obj ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( obj , types . GeneratorType ) <EOL> class _State ( threading . local ) : <EOL> """<STR_LIT>""" <EOL> current_context = None <EOL> def __init__ ( self ) : <EOL> super ( _State , self ) . __init__ ( ) <EOL> self . all_pending = set ( ) <EOL> def add_pending ( self , fut ) : <EOL> logging_debug ( '<STR_LIT>' , fut ) <EOL> self . all_pending . add ( fut ) <EOL> def remove_pending ( self , fut , status = '<STR_LIT:success>' ) : <EOL> if fut in self . all_pending : <EOL> logging_debug ( '<STR_LIT>' , status , fut ) <EOL> self . all_pending . remove ( fut ) <EOL> else : <EOL> logging_debug ( '<STR_LIT>' , status , fut ) <EOL> def clear_all_pending ( self ) : <EOL> if self . all_pending : <EOL> logging . info ( '<STR_LIT>' , self . all_pending ) <EOL> self . all_pending . clear ( ) <EOL> else : <EOL> logging_debug ( '<STR_LIT>' ) <EOL> def dump_all_pending ( self , verbose = False ) : <EOL> all = [ ] <EOL> for fut in self . all_pending : <EOL> if verbose : <EOL> line = fut . dump ( ) + ( '<STR_LIT:\n>' + '<STR_LIT:->' * <NUM_LIT> ) <EOL> else : <EOL> line = fut . dump_stack ( ) <EOL> all . append ( line ) <EOL> return '<STR_LIT:\n>' . join ( all ) <EOL> _state = _State ( ) <EOL> class Future ( object ) : <EOL> """<STR_LIT>""" <EOL> IDLE = RPC . IDLE <EOL> RUNNING = RPC . RUNNING <EOL> FINISHING = RPC . FINISHING <EOL> _geninfo = None <EOL> def __init__ ( self , info = None ) : <EOL> __ndb_debug__ = '<STR_LIT>' <EOL> self . _info = info <EOL> self . _where = utils . get_stack ( ) <EOL> self . _context = None <EOL> self . _reset ( ) <EOL> def _reset ( self ) : <EOL> self . _done = False <EOL> self . _result = None <EOL> self . _exception = None <EOL> self . _traceback = None <EOL> self . _callbacks = [ ] <EOL> _state . add_pending ( self ) <EOL> self . _next = None <EOL> def __repr__ ( self ) : <EOL> if self . _done : <EOL> if self . _exception is not None : <EOL> state = '<STR_LIT>' % ( self . _exception . __class__ . __name__ , <EOL> self . _exception ) <EOL> else : <EOL> state = '<STR_LIT>' % ( self . _result , ) <EOL> else : <EOL> state = '<STR_LIT>' <EOL> line = '<STR_LIT:?>' <EOL> for line in self . _where : <EOL> if '<STR_LIT>' not in line : <EOL> break <EOL> if self . _info : <EOL> line += '<STR_LIT>' % self . _info <EOL> if self . _geninfo : <EOL> line += '<STR_LIT>' % self . _geninfo <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , id ( self ) , line , state ) <EOL> def dump ( self ) : <EOL> return '<STR_LIT>' % ( self . dump_stack ( ) , <EOL> '<STR_LIT>' . join ( self . _where ) ) <EOL> def dump_stack ( self ) : <EOL> lines = [ ] <EOL> fut = self <EOL> while fut is not None : <EOL> lines . append ( str ( fut ) ) <EOL> fut = fut . _next <EOL> return '<STR_LIT>' . join ( lines ) <EOL> def add_callback ( self , callback , * args , ** kwds ) : <EOL> if self . _done : <EOL> eventloop . queue_call ( None , callback , * args , ** kwds ) <EOL> else : <EOL> self . _callbacks . append ( ( callback , args , kwds ) ) <EOL> def set_result ( self , result ) : <EOL> assert not self . _done <EOL> self . _result = result <EOL> self . _done = True <EOL> _state . remove_pending ( self ) <EOL> for callback , args , kwds in self . _callbacks : <EOL> eventloop . queue_call ( None , callback , * args , ** kwds ) <EOL> def set_exception ( self , exc , tb = None ) : <EOL> assert isinstance ( exc , BaseException ) <EOL> assert not self . _done <EOL> self . _exception = exc <EOL> self . _traceback = tb <EOL> self . _done = True <EOL> _state . remove_pending ( self , status = '<STR_LIT>' ) <EOL> for callback , args , kwds in self . _callbacks : <EOL> eventloop . queue_call ( None , callback , * args , ** kwds ) <EOL> def done ( self ) : <EOL> return self . _done <EOL> @ property <EOL> def state ( self ) : <EOL> if self . _done : <EOL> return self . FINISHING <EOL> else : <EOL> return self . RUNNING <EOL> def wait ( self ) : <EOL> if self . _done : <EOL> return <EOL> ev = eventloop . get_event_loop ( ) <EOL> while not self . _done : <EOL> if not ev . run1 ( ) : <EOL> logging . info ( '<STR_LIT>' , self ) <EOL> logging . info ( '<STR_LIT>' , _state . dump_all_pending ( ) ) <EOL> logging_debug ( '<STR_LIT>' , <EOL> _state . dump_all_pending ( verbose = True ) ) <EOL> self . set_exception ( RuntimeError ( '<STR_LIT>' % self ) ) <EOL> def get_exception ( self ) : <EOL> self . wait ( ) <EOL> return self . _exception <EOL> def get_traceback ( self ) : <EOL> self . wait ( ) <EOL> return self . _traceback <EOL> def check_success ( self ) : <EOL> self . wait ( ) <EOL> if self . _exception is not None : <EOL> raise self . _exception . __class__ , self . _exception , self . _traceback <EOL> def get_result ( self ) : <EOL> self . check_success ( ) <EOL> return self . _result <EOL> @ classmethod <EOL> def wait_any ( cls , futures ) : <EOL> all = set ( futures ) <EOL> ev = eventloop . get_event_loop ( ) <EOL> while all : <EOL> for f in all : <EOL> if f . state == cls . FINISHING : <EOL> return f <EOL> ev . run1 ( ) <EOL> return None <EOL> @ classmethod <EOL> def wait_all ( cls , futures ) : <EOL> all = set ( futures ) <EOL> ev = eventloop . get_event_loop ( ) <EOL> while all : <EOL> all = set ( f for f in all if f . state == cls . RUNNING ) <EOL> ev . run1 ( ) <EOL> def _help_tasklet_along ( self , gen , val = None , exc = None , tb = None ) : <EOL> info = utils . gen_info ( gen ) <EOL> __ndb_debug__ = info <EOL> try : <EOL> save_context = get_context ( ) <EOL> try : <EOL> set_context ( self . _context ) <EOL> if exc is not None : <EOL> logging_debug ( '<STR_LIT>' , <EOL> exc . __class__ . __name__ , exc , info ) <EOL> value = gen . throw ( exc . __class__ , exc , tb ) <EOL> else : <EOL> logging_debug ( '<STR_LIT>' , val , info ) <EOL> value = gen . send ( val ) <EOL> self . _context = get_context ( ) <EOL> finally : <EOL> set_context ( save_context ) <EOL> except StopIteration , err : <EOL> result = get_return_value ( err ) <EOL> logging_debug ( '<STR_LIT>' , info , result ) <EOL> self . set_result ( result ) <EOL> return <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> logging . warning ( '<STR_LIT>' , <EOL> info , err . __class__ . __name__ , err , <EOL> exc_info = ( logging . getLogger ( ) . level <= logging . INFO ) ) <EOL> self . set_exception ( err , tb ) <EOL> return <EOL> else : <EOL> logging_debug ( '<STR_LIT>' , info , value ) <EOL> if isinstance ( value , ( UserRPC , datastore_rpc . MultiRpc ) ) : <EOL> eventloop . queue_rpc ( value , self . _on_rpc_completion , value , gen ) <EOL> return <EOL> if isinstance ( value , Future ) : <EOL> assert not self . _next , self . _next <EOL> self . _next = value <EOL> self . _geninfo = utils . gen_info ( gen ) <EOL> logging_debug ( '<STR_LIT>' , self , value ) <EOL> value . add_callback ( self . _on_future_completion , value , gen ) <EOL> return <EOL> if isinstance ( value , ( tuple , list ) ) : <EOL> info = '<STR_LIT>' + utils . gen_info ( gen ) <EOL> mfut = MultiFuture ( info ) <EOL> try : <EOL> for subfuture in value : <EOL> mfut . add_dependent ( subfuture ) <EOL> mfut . complete ( ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> mfut . set_exception ( err , tb ) <EOL> mfut . add_callback ( self . _on_future_completion , mfut , gen ) <EOL> return <EOL> if is_generator ( value ) : <EOL> assert False <EOL> assert False <EOL> def _on_rpc_completion ( self , rpc , gen ) : <EOL> try : <EOL> result = rpc . get_result ( ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> self . _help_tasklet_along ( gen , exc = err , tb = tb ) <EOL> else : <EOL> self . _help_tasklet_along ( gen , result ) <EOL> def _on_future_completion ( self , future , gen ) : <EOL> if self . _next is future : <EOL> self . _next = None <EOL> self . _geninfo = None <EOL> logging_debug ( '<STR_LIT>' , self , future ) <EOL> exc = future . get_exception ( ) <EOL> if exc is not None : <EOL> self . _help_tasklet_along ( gen , exc = exc , tb = future . get_traceback ( ) ) <EOL> else : <EOL> val = future . get_result ( ) <EOL> self . _help_tasklet_along ( gen , val ) <EOL> def sleep ( dt ) : <EOL> """<STR_LIT>""" <EOL> fut = Future ( '<STR_LIT>' % dt ) <EOL> eventloop . queue_call ( dt , fut . set_result , None ) <EOL> return fut <EOL> class MultiFuture ( Future ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , info = None ) : <EOL> __ndb_debug__ = '<STR_LIT>' <EOL> self . _full = False <EOL> self . _dependents = set ( ) <EOL> self . _results = [ ] <EOL> super ( MultiFuture , self ) . __init__ ( info = info ) <EOL> def __repr__ ( self ) : <EOL> line = super ( MultiFuture , self ) . __repr__ ( ) <EOL> lines = [ line ] <EOL> for fut in self . _results : <EOL> lines . append ( fut . dump_stack ( ) . replace ( '<STR_LIT:\n>' , '<STR_LIT>' ) ) <EOL> return '<STR_LIT>' . join ( lines ) <EOL> def complete ( self ) : <EOL> assert not self . _full <EOL> self . _full = True <EOL> if not self . _dependents : <EOL> self . _finish ( ) <EOL> def set_exception ( self , exc , tb = None ) : <EOL> self . _full = True <EOL> super ( MultiFuture , self ) . set_exception ( exc , tb ) <EOL> def _finish ( self ) : <EOL> assert self . _full <EOL> assert not self . _dependents <EOL> assert not self . _done <EOL> try : <EOL> result = [ r . get_result ( ) for r in self . _results ] <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> self . set_exception ( err , tb ) <EOL> else : <EOL> self . set_result ( result ) <EOL> def putq ( self , value ) : <EOL> if isinstance ( value , Future ) : <EOL> fut = value <EOL> else : <EOL> fut = Future ( ) <EOL> fut . set_result ( value ) <EOL> self . add_dependent ( fut ) <EOL> def add_dependent ( self , fut ) : <EOL> assert isinstance ( fut , Future ) <EOL> assert not self . _full <EOL> self . _results . append ( fut ) <EOL> if fut not in self . _dependents : <EOL> self . _dependents . add ( fut ) <EOL> fut . add_callback ( self . _signal_dependent_done , fut ) <EOL> def _signal_dependent_done ( self , fut ) : <EOL> self . _dependents . remove ( fut ) <EOL> if self . _full and not self . _dependents and not self . _done : <EOL> self . _finish ( ) <EOL> class QueueFuture ( Future ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , info = None ) : <EOL> self . _full = False <EOL> self . _dependents = set ( ) <EOL> self . _completed = collections . deque ( ) <EOL> self . _waiting = collections . deque ( ) <EOL> super ( QueueFuture , self ) . __init__ ( info = info ) <EOL> def complete ( self ) : <EOL> assert not self . _full <EOL> self . _full = True <EOL> if not self . _dependents : <EOL> self . set_result ( None ) <EOL> self . _mark_finished ( ) <EOL> def set_exception ( self , exc , tb = None ) : <EOL> self . _full = True <EOL> super ( QueueFuture , self ) . set_exception ( exc , tb ) <EOL> if not self . _dependents : <EOL> self . _mark_finished ( ) <EOL> def putq ( self , value ) : <EOL> if isinstance ( value , Future ) : <EOL> fut = value <EOL> else : <EOL> fut = Future ( ) <EOL> fut . set_result ( value ) <EOL> self . add_dependent ( fut ) <EOL> def add_dependent ( self , fut ) : <EOL> assert isinstance ( fut , Future ) <EOL> assert not self . _full <EOL> if fut not in self . _dependents : <EOL> self . _dependents . add ( fut ) <EOL> fut . add_callback ( self . _signal_dependent_done , fut ) <EOL> def _signal_dependent_done ( self , fut ) : <EOL> assert fut . done ( ) <EOL> self . _dependents . remove ( fut ) <EOL> exc = fut . get_exception ( ) <EOL> tb = fut . get_traceback ( ) <EOL> val = None <EOL> if exc is None : <EOL> val = fut . get_result ( ) <EOL> if self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> self . _pass_result ( waiter , exc , tb , val ) <EOL> else : <EOL> self . _completed . append ( ( exc , tb , val ) ) <EOL> if self . _full and not self . _dependents and not self . _done : <EOL> self . set_result ( None ) <EOL> self . _mark_finished ( ) <EOL> def _mark_finished ( self ) : <EOL> assert self . _done <EOL> while self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> self . _pass_eof ( waiter ) <EOL> def getq ( self ) : <EOL> fut = Future ( ) <EOL> if self . _completed : <EOL> exc , tb , val = self . _completed . popleft ( ) <EOL> self . _pass_result ( fut , exc , tb , val ) <EOL> elif self . _full and not self . _dependents : <EOL> self . _pass_eof ( fut ) <EOL> else : <EOL> self . _waiting . append ( fut ) <EOL> return fut <EOL> def _pass_eof ( self , fut ) : <EOL> assert self . _done <EOL> exc = self . get_exception ( ) <EOL> if exc is not None : <EOL> tb = self . get_traceback ( ) <EOL> else : <EOL> exc = EOFError ( '<STR_LIT>' ) <EOL> tb = None <EOL> self . _pass_result ( fut , exc , tb , None ) <EOL> def _pass_result ( self , fut , exc , tb , val ) : <EOL> if exc is not None : <EOL> fut . set_exception ( exc , tb ) <EOL> else : <EOL> fut . set_result ( val ) <EOL> class SerialQueueFuture ( Future ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , info = None ) : <EOL> self . _full = False <EOL> self . _queue = collections . deque ( ) <EOL> self . _waiting = collections . deque ( ) <EOL> super ( SerialQueueFuture , self ) . __init__ ( info = info ) <EOL> def complete ( self ) : <EOL> assert not self . _full <EOL> self . _full = True <EOL> while self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> waiter . set_exception ( EOFError ( '<STR_LIT>' ) ) <EOL> if not self . _queue : <EOL> self . set_result ( None ) <EOL> def set_exception ( self , exc , tb = None ) : <EOL> self . _full = True <EOL> super ( SerialQueueFuture , self ) . set_exception ( exc , tb ) <EOL> while self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> waiter . set_exception ( exc , tb ) <EOL> def putq ( self , value ) : <EOL> if isinstance ( value , Future ) : <EOL> fut = value <EOL> else : <EOL> if self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> waiter . set_result ( value ) <EOL> return <EOL> fut = Future ( ) <EOL> fut . set_result ( value ) <EOL> self . add_dependent ( fut ) <EOL> def add_dependent ( self , fut ) : <EOL> assert isinstance ( fut , Future ) <EOL> assert not self . _full <EOL> if self . _waiting : <EOL> waiter = self . _waiting . popleft ( ) <EOL> fut . add_callback ( _transfer_result , fut , waiter ) <EOL> else : <EOL> self . _queue . append ( fut ) <EOL> def getq ( self ) : <EOL> if self . _queue : <EOL> fut = self . _queue . popleft ( ) <EOL> if not self . _queue and self . _full and not self . _done : <EOL> self . set_result ( None ) <EOL> else : <EOL> fut = Future ( ) <EOL> if self . _full : <EOL> assert self . _done <EOL> err = None <EOL> err = self . get_exception ( ) <EOL> if err is not None : <EOL> tb = self . get_traceback ( ) <EOL> else : <EOL> err = EOFError ( '<STR_LIT>' ) <EOL> tb = None <EOL> fut . set_exception ( err , tb ) <EOL> else : <EOL> self . _waiting . append ( fut ) <EOL> return fut <EOL> def _transfer_result ( fut1 , fut2 ) : <EOL> """<STR_LIT>""" <EOL> exc = fut1 . get_exception ( ) <EOL> if exc is not None : <EOL> tb = fut1 . get_traceback ( ) <EOL> fut2 . set_exception ( exc , tb ) <EOL> else : <EOL> val = fut1 . get_result ( ) <EOL> fut2 . set_result ( val ) <EOL> class ReducingFuture ( Future ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , reducer , info = None , batch_size = <NUM_LIT:20> ) : <EOL> self . _reducer = reducer <EOL> self . _batch_size = batch_size <EOL> self . _full = False <EOL> self . _dependents = set ( ) <EOL> self . _completed = collections . deque ( ) <EOL> self . _queue = collections . deque ( ) <EOL> super ( ReducingFuture , self ) . __init__ ( info = info ) <EOL> def complete ( self ) : <EOL> assert not self . _full <EOL> self . _full = True <EOL> if not self . _dependents : <EOL> self . _mark_finished ( ) <EOL> def set_exception ( self , exc , tb = None ) : <EOL> self . _full = True <EOL> self . _queue . clear ( ) <EOL> super ( ReducingFuture , self ) . set_exception ( exc , tb ) <EOL> def putq ( self , value ) : <EOL> if isinstance ( value , Future ) : <EOL> fut = value <EOL> else : <EOL> fut = Future ( ) <EOL> fut . set_result ( value ) <EOL> self . add_dependent ( fut ) <EOL> def add_dependent ( self , fut ) : <EOL> assert not self . _full <EOL> self . _internal_add_dependent ( fut ) <EOL> def _internal_add_dependent ( self , fut ) : <EOL> assert isinstance ( fut , Future ) <EOL> if fut not in self . _dependents : <EOL> self . _dependents . add ( fut ) <EOL> fut . add_callback ( self . _signal_dependent_done , fut ) <EOL> def _signal_dependent_done ( self , fut ) : <EOL> assert fut . done ( ) <EOL> self . _dependents . remove ( fut ) <EOL> if self . _done : <EOL> return <EOL> try : <EOL> val = fut . get_result ( ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> self . set_exception ( err , tb ) <EOL> return <EOL> self . _queue . append ( val ) <EOL> if len ( self . _queue ) >= self . _batch_size : <EOL> todo = list ( self . _queue ) <EOL> self . _queue . clear ( ) <EOL> try : <EOL> nval = self . _reducer ( todo ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> self . set_exception ( err , tb ) <EOL> return <EOL> if isinstance ( nval , Future ) : <EOL> self . _internal_add_dependent ( nval ) <EOL> else : <EOL> self . _queue . append ( nval ) <EOL> if self . _full and not self . _dependents : <EOL> self . _mark_finished ( ) <EOL> def _mark_finished ( self ) : <EOL> if not self . _queue : <EOL> self . set_result ( None ) <EOL> elif len ( self . _queue ) == <NUM_LIT:1> : <EOL> self . set_result ( self . _queue . pop ( ) ) <EOL> else : <EOL> todo = list ( self . _queue ) <EOL> self . _queue . clear ( ) <EOL> try : <EOL> nval = self . _reducer ( todo ) <EOL> except Exception , err : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> self . set_exception ( err , tb ) <EOL> return <EOL> if isinstance ( nval , Future ) : <EOL> self . _internal_add_dependent ( nval ) <EOL> else : <EOL> self . set_result ( nval ) <EOL> Return = StopIteration <EOL> def get_return_value ( err ) : <EOL> if not err . args : <EOL> result = None <EOL> elif len ( err . args ) == <NUM_LIT:1> : <EOL> result = err . args [ <NUM_LIT:0> ] <EOL> else : <EOL> result = err . args <EOL> return result <EOL> def tasklet ( func ) : <EOL> @ utils . wrapping ( func ) <EOL> def tasklet_wrapper ( * args , ** kwds ) : <EOL> fut = Future ( '<STR_LIT>' % utils . func_info ( func ) ) <EOL> fut . _context = get_context ( ) <EOL> try : <EOL> result = func ( * args , ** kwds ) <EOL> except StopIteration , err : <EOL> result = get_return_value ( err ) <EOL> if is_generator ( result ) : <EOL> eventloop . queue_call ( None , fut . _help_tasklet_along , result ) <EOL> else : <EOL> fut . set_result ( result ) <EOL> return fut <EOL> return tasklet_wrapper <EOL> def synctasklet ( func ) : <EOL> """<STR_LIT>""" <EOL> @ utils . wrapping ( func ) <EOL> def synctasklet_wrapper ( * args , ** kwds ) : <EOL> __ndb_debug__ = utils . func_info ( func ) <EOL> taskletfunc = tasklet ( func ) <EOL> return taskletfunc ( * args , ** kwds ) . get_result ( ) <EOL> return synctasklet_wrapper <EOL> _CONTEXT_KEY = '<STR_LIT>' <EOL> def get_context ( ) : <EOL> ctx = None <EOL> if os . getenv ( _CONTEXT_KEY ) : <EOL> ctx = _state . current_context <EOL> if ctx is None : <EOL> ctx = make_default_context ( ) <EOL> set_context ( ctx ) <EOL> return ctx <EOL> def make_default_context ( ) : <EOL> import context <EOL> return context . Context ( ) <EOL> def set_context ( new_context ) : <EOL> os . environ [ _CONTEXT_KEY ] = '<STR_LIT:1>' <EOL> _state . current_context = new_context </s>
