<s> import os <EOL> import os . path <EOL> import sys <EOL> import functools <EOL> import time <EOL> import imp <EOL> import code <EOL> import platform <EOL> import threading <EOL> import collections <EOL> import weakref <EOL> import gevent <EOL> import gevent . pool <EOL> import gevent . socket <EOL> import gevent . threadpool <EOL> import gevent . greenlet <EOL> import faststat <EOL> from support import context <EOL> from support import ll <EOL> ml = ll . LLogger ( ) <EOL> if hasattr ( time , "<STR_LIT>" ) : <EOL> curtime = time . perf_counter <EOL> elif platform . system ( ) == "<STR_LIT>" : <EOL> curtime = time . clock <EOL> else : <EOL> curtime = time . time <EOL> class Greenlet ( gevent . Greenlet ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , f , * a , ** kw ) : <EOL> super ( Greenlet , self ) . __init__ ( f , * a , ** kw ) <EOL> spawner = self . spawn_parent = weakref . proxy ( gevent . getcurrent ( ) ) <EOL> if not hasattr ( spawner , '<STR_LIT>' ) : <EOL> spawner . locals = { } <EOL> self . locals = spawner . locals <EOL> stack = [ ] <EOL> cur = sys . _getframe ( ) <EOL> while cur : <EOL> stack . extend ( ( cur . f_code , cur . f_lineno ) ) <EOL> cur = cur . f_back <EOL> self . stacks = ( tuple ( stack ) , ) + getattr ( spawner , '<STR_LIT>' , ( ) ) [ : <NUM_LIT:10> ] <EOL> def get_spawntree_local ( name ) : <EOL> """<STR_LIT>""" <EOL> locals = getattr ( gevent . getcurrent ( ) , '<STR_LIT>' , None ) <EOL> if locals : <EOL> return locals . get ( name ) <EOL> return None <EOL> def set_spawntree_local ( name , val ) : <EOL> """<STR_LIT>""" <EOL> cur = gevent . getcurrent ( ) <EOL> if not hasattr ( cur , '<STR_LIT>' ) : <EOL> cur . locals = { } <EOL> cur . locals [ name ] = val <EOL> def unset_spawntree_local ( name ) : <EOL> '''<STR_LIT>''' <EOL> locals = getattr ( gevent . getcurrent ( ) , '<STR_LIT>' , None ) <EOL> if locals : <EOL> del locals [ name ] <EOL> def staggered_retries ( run , * a , ** kw ) : <EOL> """<STR_LIT>""" <EOL> ctx = context . get_context ( ) <EOL> ready = gevent . event . Event ( ) <EOL> ready . clear ( ) <EOL> def callback ( source ) : <EOL> if source . successful ( ) : <EOL> ready . set ( ) <EOL> if '<STR_LIT>' in kw : <EOL> timeouts_secs = kw . pop ( '<STR_LIT>' ) <EOL> else : <EOL> timeouts_secs = [ <NUM_LIT> , <NUM_LIT:0.1> , <NUM_LIT> , <NUM_LIT> ] <EOL> if timeouts_secs [ <NUM_LIT:0> ] > <NUM_LIT:0> : <EOL> timeouts_secs . insert ( <NUM_LIT:0> , <NUM_LIT:0> ) <EOL> gs = gevent . spawn ( run , * a , ** kw ) <EOL> gs . link_value ( callback ) <EOL> running = [ gs ] <EOL> for i in range ( <NUM_LIT:1> , len ( timeouts_secs ) ) : <EOL> this_timeout = timeouts_secs [ i ] - timeouts_secs [ i - <NUM_LIT:1> ] <EOL> if ctx . dev : <EOL> this_timeout = this_timeout * <NUM_LIT> <EOL> ml . ld2 ( "<STR_LIT>" , this_timeout ) <EOL> try : <EOL> with gevent . Timeout ( this_timeout ) : <EOL> ready . wait ( ) <EOL> break <EOL> except gevent . Timeout : <EOL> ml . ld2 ( "<STR_LIT>" ) <EOL> log_rec = ctx . log . critical ( '<STR_LIT>' , run . __name__ ) <EOL> log_rec . failure ( '<STR_LIT>' , <EOL> timeout = this_timeout ) <EOL> gs = gevent . spawn ( run , * a , ** kw ) <EOL> gs . link_value ( callback ) <EOL> running . append ( gs ) <EOL> vals = [ l . value for l in running if l . successful ( ) ] <EOL> for g in running : <EOL> g . kill ( ) <EOL> if vals : <EOL> return vals [ <NUM_LIT:0> ] <EOL> else : <EOL> return None <EOL> def timed ( f ) : <EOL> '''<STR_LIT>''' <EOL> fname = os . path . basename ( f . __code__ . co_filename ) or '<STR_LIT:_>' <EOL> line_no = repr ( f . __code__ . co_firstlineno ) <EOL> name = '<STR_LIT>' . format ( f . __name__ , fname , line_no ) <EOL> @ functools . wraps ( f ) <EOL> def g ( * a , ** kw ) : <EOL> s = faststat . nanotime ( ) <EOL> r = f ( * a , ** kw ) <EOL> context . get_context ( ) . stats [ name ] . add ( ( faststat . nanotime ( ) - s ) / <NUM_LIT> ) <EOL> return r <EOL> return g <EOL> class ThreadPoolDispatcher ( object ) : <EOL> def __init__ ( self , pool , name ) : <EOL> self . pool = pool <EOL> self . name = name <EOL> def __call__ ( self , f ) : <EOL> @ functools . wraps ( f ) <EOL> def g ( * a , ** kw ) : <EOL> enqueued = curtime ( ) <EOL> ctx = context . get_context ( ) <EOL> started = [ ] <EOL> def in_thread ( * a , ** kw ) : <EOL> ml . ld3 ( "<STR_LIT>" , f . __name__ ) <EOL> started . append ( curtime ( ) ) <EOL> return f ( * a , ** kw ) <EOL> if not ctx . cpu_thread_enabled or imp . lock_held ( ) : <EOL> ret = in_thread ( * a , ** kw ) <EOL> elif in_threadpool ( ) is self . pool : <EOL> ret = in_thread ( * a , ** kw ) <EOL> else : <EOL> ctx . stats [ self . name + '<STR_LIT>' ] . add ( <NUM_LIT:1> + len ( self . pool ) ) <EOL> ret = self . pool . apply_e ( ( Exception , ) , in_thread , a , kw ) <EOL> ml . ld3 ( "<STR_LIT>" , f . __name__ , len ( pool ) ) <EOL> start = started [ <NUM_LIT:0> ] <EOL> duration = curtime ( ) - start <EOL> queued = start - enqueued <EOL> if hasattr ( ret , '<STR_LIT>' ) and callable ( ret . __len__ ) : <EOL> prsize = ret . __len__ ( ) <EOL> elif a and hasattr ( a [ <NUM_LIT:0> ] , '<STR_LIT>' ) and callable ( a [ <NUM_LIT:0> ] . __len__ ) : <EOL> prsize = a [ <NUM_LIT:0> ] . __len__ ( ) <EOL> else : <EOL> prsize = None <EOL> _queue_stats ( name , f . __name__ , queued , duration , prsize ) <EOL> return ret <EOL> g . no_defer = f <EOL> return g <EOL> def in_threadpool ( ) : <EOL> '<STR_LIT>' <EOL> frame = sys . _getframe ( ) <EOL> while frame . f_back : <EOL> frame = frame . f_back <EOL> self = frame . f_locals . get ( '<STR_LIT>' ) <EOL> if ( isinstance ( self , gevent . threadpool . ThreadPool ) and <EOL> frame . f_code is getattr ( getattr ( self . _worker , "<STR_LIT>" ) , "<STR_LIT>" ) ) : <EOL> return self <EOL> return None <EOL> class CPUThread ( object ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self ) : <EOL> self . in_q = collections . deque ( ) <EOL> self . out_q = collections . deque ( ) <EOL> self . in_async = None <EOL> self . out_async = gevent . get_hub ( ) . loop . async ( ) <EOL> self . out_q_has_data = gevent . event . Event ( ) <EOL> self . out_async . start ( self . out_q_has_data . set ) <EOL> self . worker = threading . Thread ( target = self . _run ) <EOL> self . worker . daemon = True <EOL> self . stopping = False <EOL> self . results = { } <EOL> self . worker . start ( ) <EOL> self . notifier = gevent . spawn ( self . _notify ) <EOL> def _run ( self ) : <EOL> context . get_context ( ) . thread_locals . in_cpubound_thread = True <EOL> try : <EOL> self . in_async = gevent . get_hub ( ) . loop . async ( ) <EOL> self . in_q_has_data = gevent . event . Event ( ) <EOL> self . in_async . start ( self . in_q_has_data . set ) <EOL> while not self . stopping : <EOL> if not self . in_q : <EOL> self . in_q_has_data . clear ( ) <EOL> self . in_q_has_data . wait ( ) <EOL> continue <EOL> jobid , func , args , kwargs , enqueued = self . in_q . popleft ( ) <EOL> started = curtime ( ) <EOL> try : <EOL> ret = self . results [ jobid ] = func ( * args , ** kwargs ) <EOL> except Exception as e : <EOL> ret = self . results [ jobid ] = self . _Caught ( e ) <EOL> self . out_q . append ( jobid ) <EOL> self . out_async . send ( ) <EOL> queued , duration = started - enqueued , curtime ( ) - started <EOL> size = None <EOL> if hasattr ( ret , '<STR_LIT>' ) and callable ( ret . __len__ ) : <EOL> size = len ( ret ) <EOL> _queue_stats ( '<STR_LIT>' , func . __name__ , queued , duration , size ) <EOL> except : <EOL> self . _error ( ) <EOL> def apply ( self , func , args , kwargs ) : <EOL> done = gevent . event . Event ( ) <EOL> self . in_q . append ( ( done , func , args , kwargs , curtime ( ) ) ) <EOL> context . get_context ( ) . stats [ '<STR_LIT>' ] . add ( <NUM_LIT:1> + len ( self . in_q ) ) <EOL> while not self . in_async : <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> self . in_async . send ( ) <EOL> done . wait ( ) <EOL> res = self . results [ done ] <EOL> del self . results [ done ] <EOL> if isinstance ( res , self . _Caught ) : <EOL> raise res . err <EOL> return res <EOL> def _notify ( self ) : <EOL> try : <EOL> while not self . stopping : <EOL> if not self . out_q : <EOL> self . out_q_has_data . clear ( ) <EOL> self . out_q_has_data . wait ( ) <EOL> continue <EOL> self . out_q . popleft ( ) . set ( ) <EOL> except : <EOL> self . _error ( ) <EOL> class _Caught ( object ) : <EOL> def __init__ ( self , err ) : <EOL> self . err = err <EOL> def __repr__ ( self ) : <EOL> cn = self . __class__ . __name__ <EOL> return ( "<STR_LIT>" % <EOL> ( cn , id ( self ) , len ( self . in_q ) , len ( self . out_q ) ) ) <EOL> def _error ( self ) : <EOL> import traceback <EOL> traceback . print_exc ( ) <EOL> ctx = context . get_context ( ) <EOL> tl = ctx . thread_locals <EOL> if hasattr ( tl , '<STR_LIT>' ) and tl . cpu_bound_thread is self : <EOL> del tl . cpu_bound_thread <EOL> def _queue_stats ( qname , fname , queued_ns , duration_ns , size_B = None ) : <EOL> ctx = context . get_context ( ) <EOL> fprefix = qname + '<STR_LIT:.>' + fname <EOL> ctx . stats [ fprefix + '<STR_LIT>' ] . add ( queued_ns * <NUM_LIT:1000> ) <EOL> ctx . stats [ fprefix + '<STR_LIT>' ] . add ( duration_ns * <NUM_LIT:1000> ) <EOL> ctx . stats [ qname + '<STR_LIT>' ] . add ( queued_ns * <NUM_LIT:1000> ) <EOL> ctx . stats [ qname + '<STR_LIT>' ] . add ( duration_ns * <NUM_LIT:1000> ) <EOL> if size_B is not None : <EOL> ctx . stats [ fprefix + '<STR_LIT>' ] . add ( size_B ) <EOL> if duration_ns : <EOL> ctx . stats [ fprefix + '<STR_LIT>' ] . add ( size_B / ( duration_ns * <NUM_LIT> ) ) <EOL> io_bound = ThreadPoolDispatcher ( gevent . threadpool . ThreadPool ( <NUM_LIT:10> ) , '<STR_LIT>' ) <EOL> def cpu_bound ( f , p = None ) : <EOL> '''<STR_LIT>''' <EOL> @ functools . wraps ( f ) <EOL> def g ( * a , ** kw ) : <EOL> ctx = context . get_context ( ) <EOL> if ( not ctx . cpu_thread_enabled or imp . lock_held ( ) <EOL> or getattr ( ctx . thread_locals , '<STR_LIT>' , False ) ) : <EOL> return f ( * a , ** kw ) <EOL> if not hasattr ( ctx . thread_locals , '<STR_LIT>' ) : <EOL> ctx . thread_locals . cpu_bound_thread = CPUThread ( ) <EOL> ml . ld3 ( "<STR_LIT>" , f . __name__ ) <EOL> return ctx . thread_locals . cpu_bound_thread . apply ( f , a , kw ) <EOL> g . no_defer = f <EOL> return g <EOL> def cpu_bound_if ( p ) : <EOL> '''<STR_LIT>''' <EOL> def g ( f ) : <EOL> f = cpu_bound ( f ) <EOL> @ functools . wraps ( f ) <EOL> def h ( * a , ** kw ) : <EOL> if p ( * a , ** kw ) : <EOL> return f ( * a , ** kw ) <EOL> return f . no_defer ( * a , ** kw ) <EOL> return h <EOL> return g <EOL> def close_threadpool ( ) : <EOL> tlocals = context . get_context ( ) . thread_locals <EOL> if hasattr ( tlocals , '<STR_LIT>' ) : <EOL> ml . ld2 ( "<STR_LIT>" , id ( tlocals . cpu_thread ) ) <EOL> cpu_thread = tlocals . cpu_bound_thread <EOL> cpu_thread . stopping = True <EOL> del tlocals . cpu_bound_thread <EOL> def killsock ( sock ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( sock , '<STR_LIT>' ) : <EOL> ml . ld ( "<STR_LIT>" , id ( sock ) , sock . _sock . fileno ( ) ) <EOL> else : <EOL> ml . ld ( "<STR_LIT>" , id ( sock ) ) <EOL> try : <EOL> sock . shutdown ( gevent . socket . SHUT_RDWR ) <EOL> except gevent . socket . error : <EOL> pass <EOL> except Exception as e : <EOL> log_rec = context . get_context ( ) . log . info ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> log_rec . failure ( '<STR_LIT>' , <EOL> socket = sock , exc = e ) <EOL> try : <EOL> sock . close ( ) <EOL> except gevent . socket . error : <EOL> pass <EOL> except Exception as e : <EOL> log_rec = context . get_context ( ) . log . info ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> log_rec . failure ( '<STR_LIT>' , <EOL> socket = sock , exc = e ) <EOL> PID = os . getpid ( ) <EOL> def check_fork ( fn ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( fn ) <EOL> def wrapper ( request ) : <EOL> global PID <EOL> if PID != os . getpid ( ) : <EOL> gevent . get_hub ( ) . loop . reinit ( ) <EOL> PID = os . getpid ( ) <EOL> return fn ( request ) <EOL> return wrapper <EOL> class GreenConsole ( code . InteractiveConsole ) : <EOL> @ io_bound <EOL> def raw_input ( self , prompt = "<STR_LIT>" ) : <EOL> return code . InteractiveConsole . raw_input ( self , prompt ) <EOL> def start_repl ( local = None , banner = "<STR_LIT>" ) : <EOL> gevent . spawn ( GreenConsole ( ) . interact , banner ) <EOL> def greenify ( banner = "<STR_LIT>" ) : <EOL> import __main__ <EOL> GreenConsole ( __main__ . __dict__ ) . interact ( banner ) <EOL> spawn = Greenlet . spawn <EOL> sleep = gevent . sleep <EOL> Timeout = gevent . Timeout <EOL> with_timeout = gevent . with_timeout <EOL> nanotime = faststat . nanotime </s>
