<s> __all__ = [ '<STR_LIT>' ] <EOL> import threading <EOL> import Queue <EOL> import itertools <EOL> import collections <EOL> import time <EOL> from multiprocessing import Process , cpu_count , TimeoutError <EOL> from multiprocessing . util import Finalize , debug <EOL> RUN = <NUM_LIT:0> <EOL> CLOSE = <NUM_LIT:1> <EOL> TERMINATE = <NUM_LIT:2> <EOL> job_counter = itertools . count ( ) <EOL> def mapstar ( args ) : <EOL> return map ( * args ) <EOL> class MaybeEncodingError ( Exception ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , exc , value ) : <EOL> self . exc = repr ( exc ) <EOL> self . value = repr ( value ) <EOL> super ( MaybeEncodingError , self ) . __init__ ( self . exc , self . value ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" % ( self . value , <EOL> self . exc ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % str ( self ) <EOL> def worker ( inqueue , outqueue , initializer = None , initargs = ( ) , maxtasks = None ) : <EOL> assert maxtasks is None or ( type ( maxtasks ) == int and maxtasks > <NUM_LIT:0> ) <EOL> put = outqueue . put <EOL> get = inqueue . get <EOL> if hasattr ( inqueue , '<STR_LIT>' ) : <EOL> inqueue . _writer . close ( ) <EOL> outqueue . _reader . close ( ) <EOL> if initializer is not None : <EOL> initializer ( * initargs ) <EOL> completed = <NUM_LIT:0> <EOL> while maxtasks is None or ( maxtasks and completed < maxtasks ) : <EOL> try : <EOL> task = get ( ) <EOL> except ( EOFError , IOError ) : <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> if task is None : <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> job , i , func , args , kwds = task <EOL> try : <EOL> result = ( True , func ( * args , ** kwds ) ) <EOL> except Exception , e : <EOL> result = ( False , e ) <EOL> try : <EOL> put ( ( job , i , result ) ) <EOL> except Exception as e : <EOL> wrapped = MaybeEncodingError ( e , result [ <NUM_LIT:1> ] ) <EOL> debug ( "<STR_LIT>" % ( <EOL> wrapped ) ) <EOL> put ( ( job , i , ( False , wrapped ) ) ) <EOL> completed += <NUM_LIT:1> <EOL> debug ( '<STR_LIT>' % completed ) <EOL> class Pool ( object ) : <EOL> '''<STR_LIT>''' <EOL> Process = Process <EOL> def __init__ ( self , processes = None , initializer = None , initargs = ( ) , <EOL> maxtasksperchild = None ) : <EOL> self . _setup_queues ( ) <EOL> self . _taskqueue = Queue . Queue ( ) <EOL> self . _cache = { } <EOL> self . _state = RUN <EOL> self . _maxtasksperchild = maxtasksperchild <EOL> self . _initializer = initializer <EOL> self . _initargs = initargs <EOL> if processes is None : <EOL> try : <EOL> processes = cpu_count ( ) <EOL> except NotImplementedError : <EOL> processes = <NUM_LIT:1> <EOL> if processes < <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if initializer is not None and not hasattr ( initializer , '<STR_LIT>' ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> self . _processes = processes <EOL> self . _pool = [ ] <EOL> self . _repopulate_pool ( ) <EOL> self . _worker_handler = threading . Thread ( <EOL> target = Pool . _handle_workers , <EOL> args = ( self , ) <EOL> ) <EOL> self . _worker_handler . daemon = True <EOL> self . _worker_handler . _state = RUN <EOL> self . _worker_handler . start ( ) <EOL> self . _task_handler = threading . Thread ( <EOL> target = Pool . _handle_tasks , <EOL> args = ( self . _taskqueue , self . _quick_put , self . _outqueue , self . _pool ) <EOL> ) <EOL> self . _task_handler . daemon = True <EOL> self . _task_handler . _state = RUN <EOL> self . _task_handler . start ( ) <EOL> self . _result_handler = threading . Thread ( <EOL> target = Pool . _handle_results , <EOL> args = ( self . _outqueue , self . _quick_get , self . _cache ) <EOL> ) <EOL> self . _result_handler . daemon = True <EOL> self . _result_handler . _state = RUN <EOL> self . _result_handler . start ( ) <EOL> self . _terminate = Finalize ( <EOL> self , self . _terminate_pool , <EOL> args = ( self . _taskqueue , self . _inqueue , self . _outqueue , self . _pool , <EOL> self . _worker_handler , self . _task_handler , <EOL> self . _result_handler , self . _cache ) , <EOL> exitpriority = <NUM_LIT:15> <EOL> ) <EOL> def _join_exited_workers ( self ) : <EOL> """<STR_LIT>""" <EOL> cleaned = False <EOL> for i in reversed ( range ( len ( self . _pool ) ) ) : <EOL> worker = self . _pool [ i ] <EOL> if worker . exitcode is not None : <EOL> debug ( '<STR_LIT>' % i ) <EOL> worker . join ( ) <EOL> cleaned = True <EOL> del self . _pool [ i ] <EOL> return cleaned <EOL> def _repopulate_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> for i in range ( self . _processes - len ( self . _pool ) ) : <EOL> w = self . Process ( target = worker , <EOL> args = ( self . _inqueue , self . _outqueue , <EOL> self . _initializer , <EOL> self . _initargs , self . _maxtasksperchild ) <EOL> ) <EOL> self . _pool . append ( w ) <EOL> w . name = w . name . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> w . daemon = True <EOL> w . start ( ) <EOL> debug ( '<STR_LIT>' ) <EOL> def _maintain_pool ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _join_exited_workers ( ) : <EOL> self . _repopulate_pool ( ) <EOL> def _setup_queues ( self ) : <EOL> from . queues import SimpleQueue <EOL> self . _inqueue = SimpleQueue ( ) <EOL> self . _outqueue = SimpleQueue ( ) <EOL> self . _quick_put = self . _inqueue . _writer . send <EOL> self . _quick_get = self . _outqueue . _reader . recv <EOL> def apply ( self , func , args = ( ) , kwds = { } ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> return self . apply_async ( func , args , kwds ) . get ( ) <EOL> def map ( self , func , iterable , chunksize = None ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> return self . map_async ( func , iterable , chunksize ) . get ( ) <EOL> def imap ( self , func , iterable , chunksize = <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> if chunksize == <NUM_LIT:1> : <EOL> result = IMapIterator ( self . _cache ) <EOL> self . _taskqueue . put ( ( ( ( result . _job , i , func , ( x , ) , { } ) <EOL> for i , x in enumerate ( iterable ) ) , result . _set_length ) ) <EOL> return result <EOL> else : <EOL> assert chunksize > <NUM_LIT:1> <EOL> task_batches = Pool . _get_tasks ( func , iterable , chunksize ) <EOL> result = IMapIterator ( self . _cache ) <EOL> self . _taskqueue . put ( ( ( ( result . _job , i , mapstar , ( x , ) , { } ) <EOL> for i , x in enumerate ( task_batches ) ) , result . _set_length ) ) <EOL> return ( item for chunk in result for item in chunk ) <EOL> def imap_unordered ( self , func , iterable , chunksize = <NUM_LIT:1> ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> if chunksize == <NUM_LIT:1> : <EOL> result = IMapUnorderedIterator ( self . _cache ) <EOL> self . _taskqueue . put ( ( ( ( result . _job , i , func , ( x , ) , { } ) <EOL> for i , x in enumerate ( iterable ) ) , result . _set_length ) ) <EOL> return result <EOL> else : <EOL> assert chunksize > <NUM_LIT:1> <EOL> task_batches = Pool . _get_tasks ( func , iterable , chunksize ) <EOL> result = IMapUnorderedIterator ( self . _cache ) <EOL> self . _taskqueue . put ( ( ( ( result . _job , i , mapstar , ( x , ) , { } ) <EOL> for i , x in enumerate ( task_batches ) ) , result . _set_length ) ) <EOL> return ( item for chunk in result for item in chunk ) <EOL> def apply_async ( self , func , args = ( ) , kwds = { } , callback = None ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> result = ApplyResult ( self . _cache , callback ) <EOL> self . _taskqueue . put ( ( [ ( result . _job , None , func , args , kwds ) ] , None ) ) <EOL> return result <EOL> def map_async ( self , func , iterable , chunksize = None , callback = None ) : <EOL> '''<STR_LIT>''' <EOL> assert self . _state == RUN <EOL> if not hasattr ( iterable , '<STR_LIT>' ) : <EOL> iterable = list ( iterable ) <EOL> if chunksize is None : <EOL> chunksize , extra = divmod ( len ( iterable ) , len ( self . _pool ) * <NUM_LIT:4> ) <EOL> if extra : <EOL> chunksize += <NUM_LIT:1> <EOL> if len ( iterable ) == <NUM_LIT:0> : <EOL> chunksize = <NUM_LIT:0> <EOL> task_batches = Pool . _get_tasks ( func , iterable , chunksize ) <EOL> result = MapResult ( self . _cache , chunksize , len ( iterable ) , callback ) <EOL> self . _taskqueue . put ( ( ( ( result . _job , i , mapstar , ( x , ) , { } ) <EOL> for i , x in enumerate ( task_batches ) ) , None ) ) <EOL> return result <EOL> @ staticmethod <EOL> def _handle_workers ( pool ) : <EOL> thread = threading . current_thread ( ) <EOL> while thread . _state == RUN or ( pool . _cache and thread . _state != TERMINATE ) : <EOL> pool . _maintain_pool ( ) <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> pool . _taskqueue . put ( None ) <EOL> debug ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _handle_tasks ( taskqueue , put , outqueue , pool ) : <EOL> thread = threading . current_thread ( ) <EOL> for taskseq , set_length in iter ( taskqueue . get , None ) : <EOL> i = - <NUM_LIT:1> <EOL> for i , task in enumerate ( taskseq ) : <EOL> if thread . _state : <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> try : <EOL> put ( task ) <EOL> except IOError : <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> else : <EOL> if set_length : <EOL> debug ( '<STR_LIT>' ) <EOL> set_length ( i + <NUM_LIT:1> ) <EOL> continue <EOL> break <EOL> else : <EOL> debug ( '<STR_LIT>' ) <EOL> try : <EOL> debug ( '<STR_LIT>' ) <EOL> outqueue . put ( None ) <EOL> debug ( '<STR_LIT>' ) <EOL> for p in pool : <EOL> put ( None ) <EOL> except IOError : <EOL> debug ( '<STR_LIT>' ) <EOL> debug ( '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _handle_results ( outqueue , get , cache ) : <EOL> thread = threading . current_thread ( ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> task = get ( ) <EOL> except ( IOError , EOFError ) : <EOL> debug ( '<STR_LIT>' ) <EOL> return <EOL> if thread . _state : <EOL> assert thread . _state == TERMINATE <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> if task is None : <EOL> debug ( '<STR_LIT>' ) <EOL> break <EOL> job , i , obj = task <EOL> try : <EOL> cache [ job ] . _set ( i , obj ) <EOL> except KeyError : <EOL> pass <EOL> while cache and thread . _state != TERMINATE : <EOL> try : <EOL> task = get ( ) <EOL> except ( IOError , EOFError ) : <EOL> debug ( '<STR_LIT>' ) <EOL> return <EOL> if task is None : <EOL> debug ( '<STR_LIT>' ) <EOL> continue <EOL> job , i , obj = task <EOL> try : <EOL> cache [ job ] . _set ( i , obj ) <EOL> except KeyError : <EOL> pass <EOL> if hasattr ( outqueue , '<STR_LIT>' ) : <EOL> debug ( '<STR_LIT>' ) <EOL> try : <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> if not outqueue . _reader . poll ( ) : <EOL> break <EOL> get ( ) <EOL> except ( IOError , EOFError ) : <EOL> pass <EOL> debug ( '<STR_LIT>' , <EOL> len ( cache ) , thread . _state ) <EOL> @ staticmethod <EOL> def _get_tasks ( func , it , size ) : <EOL> it = iter ( it ) <EOL> while <NUM_LIT:1> : <EOL> x = tuple ( itertools . islice ( it , size ) ) <EOL> if not x : <EOL> return <EOL> yield ( func , x ) <EOL> def __reduce__ ( self ) : <EOL> raise NotImplementedError ( <EOL> '<STR_LIT>' <EOL> ) <EOL> def close ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> if self . _state == RUN : <EOL> self . _state = CLOSE <EOL> self . _worker_handler . _state = CLOSE <EOL> def terminate ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> self . _state = TERMINATE <EOL> self . _worker_handler . _state = TERMINATE <EOL> self . _terminate ( ) <EOL> def join ( self ) : <EOL> debug ( '<STR_LIT>' ) <EOL> assert self . _state in ( CLOSE , TERMINATE ) <EOL> self . _worker_handler . join ( ) <EOL> self . _task_handler . join ( ) <EOL> self . _result_handler . join ( ) <EOL> for p in self . _pool : <EOL> p . join ( ) <EOL> @ staticmethod <EOL> def _help_stuff_finish ( inqueue , task_handler , size ) : <EOL> debug ( '<STR_LIT>' ) <EOL> inqueue . _rlock . acquire ( ) <EOL> while task_handler . is_alive ( ) and inqueue . _reader . poll ( ) : <EOL> inqueue . _reader . recv ( ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> @ classmethod <EOL> def _terminate_pool ( cls , taskqueue , inqueue , outqueue , pool , <EOL> worker_handler , task_handler , result_handler , cache ) : <EOL> debug ( '<STR_LIT>' ) <EOL> worker_handler . _state = TERMINATE <EOL> task_handler . _state = TERMINATE <EOL> debug ( '<STR_LIT>' ) <EOL> cls . _help_stuff_finish ( inqueue , task_handler , len ( pool ) ) <EOL> assert result_handler . is_alive ( ) or len ( cache ) == <NUM_LIT:0> <EOL> result_handler . _state = TERMINATE <EOL> outqueue . put ( None ) <EOL> debug ( '<STR_LIT>' ) <EOL> if threading . current_thread ( ) is not worker_handler : <EOL> worker_handler . join ( <NUM_LIT> ) <EOL> if pool and hasattr ( pool [ <NUM_LIT:0> ] , '<STR_LIT>' ) : <EOL> debug ( '<STR_LIT>' ) <EOL> for p in pool : <EOL> if p . exitcode is None : <EOL> p . terminate ( ) <EOL> debug ( '<STR_LIT>' ) <EOL> if threading . current_thread ( ) is not task_handler : <EOL> task_handler . join ( <NUM_LIT> ) <EOL> debug ( '<STR_LIT>' ) <EOL> if threading . current_thread ( ) is not result_handler : <EOL> result_handler . join ( <NUM_LIT> ) <EOL> if pool and hasattr ( pool [ <NUM_LIT:0> ] , '<STR_LIT>' ) : <EOL> debug ( '<STR_LIT>' ) <EOL> for p in pool : <EOL> if p . is_alive ( ) : <EOL> debug ( '<STR_LIT>' % p . pid ) <EOL> p . join ( ) <EOL> class ApplyResult ( object ) : <EOL> def __init__ ( self , cache , callback ) : <EOL> self . _cond = threading . Condition ( threading . Lock ( ) ) <EOL> self . _job = job_counter . next ( ) <EOL> self . _cache = cache <EOL> self . _ready = False <EOL> self . _callback = callback <EOL> cache [ self . _job ] = self <EOL> def ready ( self ) : <EOL> return self . _ready <EOL> def successful ( self ) : <EOL> assert self . _ready <EOL> return self . _success <EOL> def wait ( self , timeout = None ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> if not self . _ready : <EOL> self . _cond . wait ( timeout ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> def get ( self , timeout = None ) : <EOL> self . wait ( timeout ) <EOL> if not self . _ready : <EOL> raise TimeoutError <EOL> if self . _success : <EOL> return self . _value <EOL> else : <EOL> raise self . _value <EOL> def _set ( self , i , obj ) : <EOL> self . _success , self . _value = obj <EOL> if self . _callback and self . _success : <EOL> self . _callback ( self . _value ) <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> self . _ready = True <EOL> self . _cond . notify ( ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> del self . _cache [ self . _job ] <EOL> AsyncResult = ApplyResult <EOL> class MapResult ( ApplyResult ) : <EOL> def __init__ ( self , cache , chunksize , length , callback ) : <EOL> ApplyResult . __init__ ( self , cache , callback ) <EOL> self . _success = True <EOL> self . _value = [ None ] * length <EOL> self . _chunksize = chunksize <EOL> if chunksize <= <NUM_LIT:0> : <EOL> self . _number_left = <NUM_LIT:0> <EOL> self . _ready = True <EOL> del cache [ self . _job ] <EOL> else : <EOL> self . _number_left = length // chunksize + bool ( length % chunksize ) <EOL> def _set ( self , i , success_result ) : <EOL> success , result = success_result <EOL> if success : <EOL> self . _value [ i * self . _chunksize : ( i + <NUM_LIT:1> ) * self . _chunksize ] = result <EOL> self . _number_left -= <NUM_LIT:1> <EOL> if self . _number_left == <NUM_LIT:0> : <EOL> if self . _callback : <EOL> self . _callback ( self . _value ) <EOL> del self . _cache [ self . _job ] <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> self . _ready = True <EOL> self . _cond . notify ( ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> else : <EOL> self . _success = False <EOL> self . _value = result <EOL> del self . _cache [ self . _job ] <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> self . _ready = True <EOL> self . _cond . notify ( ) <EOL> finally : <EOL> self . _cond . release ( ) <EOL> class IMapIterator ( object ) : <EOL> def __init__ ( self , cache ) : <EOL> self . _cond = threading . Condition ( threading . Lock ( ) ) <EOL> self . _job = job_counter . next ( ) <EOL> self . _cache = cache <EOL> self . _items = collections . deque ( ) <EOL> self . _index = <NUM_LIT:0> <EOL> self . _length = None <EOL> self . _unsorted = { } <EOL> cache [ self . _job ] = self <EOL> def __iter__ ( self ) : <EOL> return self <EOL> def next ( self , timeout = None ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> try : <EOL> item = self . _items . popleft ( ) <EOL> except IndexError : <EOL> if self . _index == self . _length : <EOL> raise StopIteration <EOL> self . _cond . wait ( timeout ) <EOL> try : <EOL> item = self . _items . popleft ( ) <EOL> except IndexError : <EOL> if self . _index == self . _length : <EOL> raise StopIteration <EOL> raise TimeoutError <EOL> finally : <EOL> self . _cond . release ( ) <EOL> success , value = item <EOL> if success : <EOL> return value <EOL> raise value <EOL> __next__ = next <EOL> def _set ( self , i , obj ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> if self . _index == i : <EOL> self . _items . append ( obj ) <EOL> self . _index += <NUM_LIT:1> <EOL> while self . _index in self . _unsorted : <EOL> obj = self . _unsorted . pop ( self . _index ) <EOL> self . _items . append ( obj ) <EOL> self . _index += <NUM_LIT:1> <EOL> self . _cond . notify ( ) <EOL> else : <EOL> self . _unsorted [ i ] = obj <EOL> if self . _index == self . _length : <EOL> del self . _cache [ self . _job ] <EOL> finally : <EOL> self . _cond . release ( ) <EOL> def _set_length ( self , length ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> self . _length = length <EOL> if self . _index == self . _length : <EOL> self . _cond . notify ( ) <EOL> del self . _cache [ self . _job ] <EOL> finally : <EOL> self . _cond . release ( ) <EOL> class IMapUnorderedIterator ( IMapIterator ) : <EOL> def _set ( self , i , obj ) : <EOL> self . _cond . acquire ( ) <EOL> try : <EOL> self . _items . append ( obj ) <EOL> self . _index += <NUM_LIT:1> <EOL> self . _cond . notify ( ) <EOL> if self . _index == self . _length : <EOL> del self . _cache [ self . _job ] <EOL> finally : <EOL> self . _cond . release ( ) <EOL> class ThreadPool ( Pool ) : <EOL> from . dummy import Process <EOL> def __init__ ( self , processes = None , initializer = None , initargs = ( ) ) : <EOL> Pool . __init__ ( self , processes , initializer , initargs ) <EOL> def _setup_queues ( self ) : <EOL> self . _inqueue = Queue . Queue ( ) <EOL> self . _outqueue = Queue . Queue ( ) <EOL> self . _quick_put = self . _inqueue . put <EOL> self . _quick_get = self . _outqueue . get <EOL> @ staticmethod <EOL> def _help_stuff_finish ( inqueue , task_handler , size ) : <EOL> inqueue . not_empty . acquire ( ) <EOL> try : <EOL> inqueue . queue . clear ( ) <EOL> inqueue . queue . extend ( [ None ] * size ) <EOL> inqueue . not_empty . notify_all ( ) <EOL> finally : <EOL> inqueue . not_empty . release ( ) </s>
