<s> """<STR_LIT>""" <EOL> import logging <EOL> import Queue <EOL> import sys <EOL> import threading <EOL> import time <EOL> import traceback <EOL> from google . appengine . tools . requeue import ReQueue <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> _THREAD_SHOULD_EXIT = '<STR_LIT>' <EOL> INITIAL_BACKOFF = <NUM_LIT:1.0> <EOL> BACKOFF_FACTOR = <NUM_LIT> <EOL> class Error ( Exception ) : <EOL> """<STR_LIT>""" <EOL> class WorkItemError ( Error ) : <EOL> """<STR_LIT>""" <EOL> class RetryException ( Error ) : <EOL> """<STR_LIT>""" <EOL> def InterruptibleSleep ( sleep_time ) : <EOL> """<STR_LIT>""" <EOL> slept = <NUM_LIT:0.0> <EOL> epsilon = <NUM_LIT> <EOL> thread = threading . currentThread ( ) <EOL> while slept < sleep_time - epsilon : <EOL> remaining = sleep_time - slept <EOL> this_sleep_time = min ( remaining , <NUM_LIT> ) <EOL> time . sleep ( this_sleep_time ) <EOL> slept += this_sleep_time <EOL> if thread . exit_flag : <EOL> return <EOL> class WorkerThread ( threading . Thread ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , thread_pool , thread_gate , name = None ) : <EOL> """<STR_LIT>""" <EOL> threading . Thread . __init__ ( self ) <EOL> self . setDaemon ( True ) <EOL> self . exit_flag = False <EOL> self . __error = None <EOL> self . __traceback = None <EOL> self . __thread_pool = thread_pool <EOL> self . __work_queue = thread_pool . requeue <EOL> self . __thread_gate = thread_gate <EOL> if not name : <EOL> self . __name = '<STR_LIT>' + self . __class__ . __name__ <EOL> else : <EOL> self . __name = name <EOL> def run ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> try : <EOL> self . WorkOnItems ( ) <EOL> except : <EOL> self . SetError ( ) <EOL> logger . debug ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> def SetError ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . __error : <EOL> exc_info = sys . exc_info ( ) <EOL> self . __error = exc_info [ <NUM_LIT:1> ] <EOL> self . __traceback = exc_info [ <NUM_LIT:2> ] <EOL> logger . exception ( '<STR_LIT>' , self . getName ( ) , self . __class__ . __name__ ) <EOL> def WorkOnItems ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . exit_flag : <EOL> item = None <EOL> self . __thread_gate . StartWork ( ) <EOL> try : <EOL> status , instruction = WorkItem . FAILURE , ThreadGate . DECREASE <EOL> try : <EOL> if self . exit_flag : <EOL> instruction = ThreadGate . HOLD <EOL> break <EOL> try : <EOL> item = self . __work_queue . get ( block = True , timeout = <NUM_LIT:1.0> ) <EOL> except Queue . Empty : <EOL> instruction = ThreadGate . HOLD <EOL> continue <EOL> if item == _THREAD_SHOULD_EXIT or self . exit_flag : <EOL> status , instruction = WorkItem . SUCCESS , ThreadGate . HOLD <EOL> break <EOL> logger . debug ( '<STR_LIT>' , self . getName ( ) , item ) <EOL> status , instruction = item . PerformWork ( self . __thread_pool ) <EOL> except RetryException : <EOL> status , instruction = WorkItem . RETRY , ThreadGate . HOLD <EOL> except : <EOL> self . SetError ( ) <EOL> raise <EOL> finally : <EOL> try : <EOL> if item : <EOL> if status == WorkItem . SUCCESS : <EOL> self . __work_queue . task_done ( ) <EOL> elif status == WorkItem . RETRY : <EOL> try : <EOL> self . __work_queue . reput ( item , block = False ) <EOL> except Queue . Full : <EOL> logger . error ( '<STR_LIT>' , self . getName ( ) ) <EOL> raise Error ( '<STR_LIT>' ) <EOL> else : <EOL> if not self . __error : <EOL> if item . error : <EOL> self . __error = item . error <EOL> self . __traceback = item . traceback <EOL> else : <EOL> self . __error = WorkItemError ( <EOL> '<STR_LIT>' % item ) <EOL> raise self . __error <EOL> finally : <EOL> self . __thread_gate . FinishWork ( instruction = instruction ) <EOL> def CheckError ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . __error : <EOL> logger . error ( '<STR_LIT>' , self . getName ( ) , self . __error ) <EOL> if self . __traceback : <EOL> logger . debug ( '<STR_LIT:%s>' , '<STR_LIT>' . join ( traceback . format_exception ( <EOL> self . __error . __class__ , <EOL> self . __error , <EOL> self . __traceback ) ) ) <EOL> def __str__ ( self ) : <EOL> return self . __name <EOL> class AdaptiveThreadPool ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , <EOL> num_threads , <EOL> queue_size = None , <EOL> base_thread_name = None , <EOL> worker_thread_factory = WorkerThread , <EOL> queue_factory = Queue . Queue ) : <EOL> """<STR_LIT>""" <EOL> if queue_size is None : <EOL> queue_size = num_threads <EOL> self . requeue = ReQueue ( queue_size , queue_factory = queue_factory ) <EOL> self . __thread_gate = ThreadGate ( num_threads ) <EOL> self . __num_threads = num_threads <EOL> self . __threads = [ ] <EOL> for i in xrange ( num_threads ) : <EOL> thread = worker_thread_factory ( self , self . __thread_gate ) <EOL> if base_thread_name : <EOL> base = base_thread_name <EOL> else : <EOL> base = thread . __class__ . __name__ <EOL> thread . name = '<STR_LIT>' % ( base , i ) <EOL> self . __threads . append ( thread ) <EOL> thread . start ( ) <EOL> def num_threads ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __num_threads <EOL> def Threads ( self ) : <EOL> """<STR_LIT>""" <EOL> for thread in self . __threads : <EOL> yield thread <EOL> def SubmitItem ( self , item , block = True , timeout = <NUM_LIT:0.0> ) : <EOL> """<STR_LIT>""" <EOL> self . requeue . put ( item , block = block , timeout = timeout ) <EOL> def QueuedItemCount ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . requeue . qsize ( ) <EOL> def Shutdown ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . requeue . empty ( ) : <EOL> try : <EOL> unused_item = self . requeue . get_nowait ( ) <EOL> self . requeue . task_done ( ) <EOL> except Queue . Empty : <EOL> pass <EOL> for thread in self . __threads : <EOL> thread . exit_flag = True <EOL> self . requeue . put ( _THREAD_SHOULD_EXIT ) <EOL> self . __thread_gate . EnableAllThreads ( ) <EOL> def Wait ( self ) : <EOL> """<STR_LIT>""" <EOL> self . requeue . join ( ) <EOL> def JoinThreads ( self ) : <EOL> """<STR_LIT>""" <EOL> for thread in self . __threads : <EOL> logger . debug ( '<STR_LIT>' % str ( thread ) ) <EOL> thread . join ( ) <EOL> def CheckErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> for thread in self . __threads : <EOL> thread . CheckError ( ) <EOL> class ThreadGate ( object ) : <EOL> """<STR_LIT>""" <EOL> INCREASE = '<STR_LIT>' <EOL> HOLD = '<STR_LIT>' <EOL> DECREASE = '<STR_LIT>' <EOL> def __init__ ( self , <EOL> num_threads , <EOL> sleep = InterruptibleSleep ) : <EOL> """<STR_LIT>""" <EOL> self . __enabled_count = <NUM_LIT:1> <EOL> self . __lock = threading . Lock ( ) <EOL> self . __thread_semaphore = threading . Semaphore ( self . __enabled_count ) <EOL> self . __num_threads = num_threads <EOL> self . __backoff_time = <NUM_LIT:0> <EOL> self . __sleep = sleep <EOL> def num_threads ( self ) : <EOL> return self . __num_threads <EOL> def EnableThread ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __lock . acquire ( ) <EOL> try : <EOL> self . __enabled_count += <NUM_LIT:1> <EOL> finally : <EOL> self . __lock . release ( ) <EOL> self . __thread_semaphore . release ( ) <EOL> def EnableAllThreads ( self ) : <EOL> """<STR_LIT>""" <EOL> for unused_idx in xrange ( self . __num_threads - self . __enabled_count ) : <EOL> self . EnableThread ( ) <EOL> def StartWork ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __thread_semaphore . acquire ( ) <EOL> if self . __backoff_time > <NUM_LIT:0.0> : <EOL> if not threading . currentThread ( ) . exit_flag : <EOL> logger . info ( '<STR_LIT>' , <EOL> threading . currentThread ( ) . getName ( ) , <EOL> self . __backoff_time ) <EOL> self . __sleep ( self . __backoff_time ) <EOL> def FinishWork ( self , instruction = None ) : <EOL> """<STR_LIT>""" <EOL> if not instruction or instruction == ThreadGate . HOLD : <EOL> self . __thread_semaphore . release ( ) <EOL> elif instruction == ThreadGate . INCREASE : <EOL> if self . __backoff_time > <NUM_LIT:0.0> : <EOL> logger . info ( '<STR_LIT>' ) <EOL> self . __backoff_time = <NUM_LIT:0.0> <EOL> do_enable = False <EOL> self . __lock . acquire ( ) <EOL> try : <EOL> if self . __num_threads > self . __enabled_count : <EOL> do_enable = True <EOL> self . __enabled_count += <NUM_LIT:1> <EOL> finally : <EOL> self . __lock . release ( ) <EOL> if do_enable : <EOL> logger . debug ( '<STR_LIT>' , <EOL> self . __enabled_count ) <EOL> self . __thread_semaphore . release ( ) <EOL> self . __thread_semaphore . release ( ) <EOL> elif instruction == ThreadGate . DECREASE : <EOL> do_disable = False <EOL> self . __lock . acquire ( ) <EOL> try : <EOL> if self . __enabled_count > <NUM_LIT:1> : <EOL> do_disable = True <EOL> self . __enabled_count -= <NUM_LIT:1> <EOL> else : <EOL> if self . __backoff_time == <NUM_LIT:0.0> : <EOL> self . __backoff_time = INITIAL_BACKOFF <EOL> else : <EOL> self . __backoff_time *= BACKOFF_FACTOR <EOL> finally : <EOL> self . __lock . release ( ) <EOL> if do_disable : <EOL> logger . debug ( '<STR_LIT>' , <EOL> self . __enabled_count ) <EOL> else : <EOL> self . __thread_semaphore . release ( ) <EOL> class WorkItem ( object ) : <EOL> """<STR_LIT>""" <EOL> SUCCESS = '<STR_LIT:success>' <EOL> RETRY = '<STR_LIT>' <EOL> FAILURE = '<STR_LIT>' <EOL> def __init__ ( self , name ) : <EOL> self . __name = name <EOL> def PerformWork ( self , thread_pool ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def __str__ ( self ) : <EOL> return self . __name </s>
