<s> """<STR_LIT>""" <EOL> import logging <EOL> import gevent <EOL> from gevent import pool <EOL> import time <EOL> from multiprocessing import cpu_count <EOL> log = logging . getLogger ( __name__ ) <EOL> class CallbackHandler ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , worker_func , task_count , worker_count = - <NUM_LIT:1> ) : <EOL> if worker_count == - <NUM_LIT:1> : <EOL> worker_count = cpu_count ( ) <EOL> self . worker_pool = pool . Pool ( size = worker_count ) <EOL> self . worker_pool_closed = False <EOL> self . task_count = task_count <EOL> self . worker_func = worker_func <EOL> def run_callback ( self , * args , ** kwargs ) : <EOL> blocking = kwargs . pop ( '<STR_LIT>' , False ) <EOL> if not self . worker_pool_closed : <EOL> log . debug ( '<STR_LIT>' ) <EOL> worker_params = dict ( args = args , <EOL> kwargs = kwargs ) <EOL> callback_launch_greenlet = gevent . spawn ( self . _spawn_callback , <EOL> worker_params , <EOL> blocking = blocking ) <EOL> if blocking : <EOL> callback_launch_greenlet . join ( ) <EOL> def skip ( self ) : <EOL> log . debug ( '<STR_LIT>' ) <EOL> self . _dec_task_count_skipped ( ) <EOL> def join ( self ) : <EOL> log . debug ( '<STR_LIT>' ) <EOL> last_task_count = self . task_count <EOL> last_task_elapsed = <NUM_LIT:0.0> <EOL> task_wait_timed_out = False <EOL> while self . task_count > <NUM_LIT:0> : <EOL> if last_task_count > self . task_count : <EOL> last_task_count = self . task_count <EOL> last_task_elapsed = <NUM_LIT:0.0> <EOL> time . sleep ( <NUM_LIT> ) <EOL> last_task_elapsed = last_task_elapsed + <NUM_LIT> <EOL> if last_task_elapsed > <NUM_LIT> : <EOL> task_wait_timed_out = True <EOL> break <EOL> self . worker_pool_closed = True <EOL> if task_wait_timed_out : <EOL> self . worker_pool . kill ( ) <EOL> self . worker_pool . join ( ) <EOL> if task_wait_timed_out : <EOL> log . debug ( '<STR_LIT>' ) <EOL> else : <EOL> log . debug ( '<STR_LIT>' ) <EOL> def terminate ( self ) : <EOL> log . debug ( '<STR_LIT>' ) <EOL> self . worker_pool_closed = True <EOL> self . worker_pool . kill ( ) <EOL> self . worker_pool . join ( ) <EOL> log . debug ( '<STR_LIT>' ) <EOL> def _spawn_callback ( self , worker_params , blocking = False ) : <EOL> self . worker_pool . wait_available ( ) <EOL> if not self . worker_pool_closed : <EOL> callback_greenlet = self . worker_pool . spawn ( self . _callback_func , worker_params ) <EOL> if blocking : <EOL> callback_greenlet . join ( ) <EOL> def _callback_func ( self , worker_params ) : <EOL> self . worker_func ( * worker_params [ '<STR_LIT:args>' ] , ** worker_params [ '<STR_LIT>' ] ) <EOL> self . _dec_task_count_completed ( ) <EOL> def _dec_task_count_completed ( self ) : <EOL> self . task_count = self . task_count - <NUM_LIT:1> <EOL> log . debug ( '<STR_LIT>' , self . task_count ) <EOL> def _dec_task_count_skipped ( self ) : <EOL> self . task_count = self . task_count - <NUM_LIT:1> <EOL> log . debug ( '<STR_LIT>' , self . task_count ) </s>
