<s> """<STR_LIT>""" <EOL> import itertools <EOL> from multiprocessing import cpu_count , Pool , Process , Manager , Lock <EOL> from multiprocessing . pool import IMapUnorderedIterator , IMapIterator <EOL> try : <EOL> import cPickle <EOL> except ImportError : <EOL> import pickle as cPickle <EOL> import sys <EOL> GOOGLE = <NUM_LIT> <EOL> def _do_work_off_queue ( lock , in_q , func , out_q , sep ) : <EOL> while True : <EOL> x = in_q . get ( ) <EOL> if x is None : <EOL> out_q . put ( x ) <EOL> return <EOL> result = func ( x ) <EOL> out_q . put ( str ( result ) + sep ) <EOL> def _write_to_output ( out_q , stream , n_jobs ) : <EOL> ends_seen = <NUM_LIT:0> <EOL> while True : <EOL> x = out_q . get ( ) <EOL> if not x : <EOL> ends_seen += <NUM_LIT:1> <EOL> if ends_seen == n_jobs : <EOL> stream . flush ( ) <EOL> return <EOL> else : <EOL> continue <EOL> stream . write ( x ) <EOL> def parallel_apply ( func , iterable , n_jobs , sep = '<STR_LIT:\n>' , out_stream = sys . stdout ) : <EOL> """<STR_LIT>""" <EOL> if n_jobs == <NUM_LIT:1> : <EOL> for each in iterable : <EOL> out_stream . write ( str ( func ( each ) ) + sep ) <EOL> out_stream . flush ( ) <EOL> return <EOL> manager = Manager ( ) <EOL> in_q = manager . Queue ( maxsize = <NUM_LIT:2> * n_jobs ) <EOL> out_q = manager . Queue ( maxsize = <NUM_LIT:2> * n_jobs ) <EOL> lock = Lock ( ) <EOL> pool = [ ] <EOL> for i in xrange ( n_jobs ) : <EOL> p = Process ( target = _do_work_off_queue , <EOL> args = ( lock , in_q , func , out_q , sep ) ) <EOL> p . start ( ) <EOL> pool . append ( p ) <EOL> out_p = Process ( target = _write_to_output , <EOL> args = ( out_q , out_stream , n_jobs ) ) <EOL> out_p . start ( ) <EOL> iters = itertools . chain ( iterable , ( None , ) * n_jobs ) <EOL> for each in iters : <EOL> in_q . put ( each ) <EOL> for p in pool : <EOL> p . join ( ) <EOL> out_p . join ( ) <EOL> def imap_easy ( func , iterable , n_jobs , chunksize , ordered = True ) : <EOL> """<STR_LIT>""" <EOL> n_jobs = _n_jobs_wrap ( n_jobs ) <EOL> if n_jobs == <NUM_LIT:1> : <EOL> results_iter = itertools . imap ( func , iterable ) <EOL> else : <EOL> _trypickle ( func ) <EOL> pool = Pool ( n_jobs ) <EOL> if ordered : <EOL> results_iter = pool . imap ( func , iterable , chunksize = chunksize ) <EOL> else : <EOL> results_iter = pool . imap_unordered ( <EOL> func , iterable , chunksize = chunksize ) <EOL> return results_iter <EOL> def map_easy ( func , iterable , n_jobs ) : <EOL> """<STR_LIT>""" <EOL> n_jobs = _n_jobs_wrap ( n_jobs ) <EOL> if n_jobs == <NUM_LIT:1> : <EOL> return map ( func , iterable ) <EOL> else : <EOL> _trypickle ( func ) <EOL> pool = Pool ( n_jobs ) <EOL> return pool . map_async ( func , iterable ) . get ( GOOGLE ) <EOL> def map_easy_padded_blocks ( func , iterable , n_jobs , pad , blocksize = None ) : <EOL> """<STR_LIT>""" <EOL> mylist = list ( iterable ) <EOL> max_blocksize = len ( mylist ) - pad - <NUM_LIT:1> <EOL> if blocksize is None : <EOL> blocksize = min ( max_blocksize , <NUM_LIT:100> * pad ) <EOL> assert pad + blocksize < len ( mylist ) <EOL> block_idx , pads_used = _get_split_idx ( len ( mylist ) , blocksize , pad = pad ) <EOL> block_iter = ( mylist [ start : end ] for start , end in block_idx ) <EOL> processed_blocks = map_easy ( func , block_iter , n_jobs ) <EOL> result = [ ] <EOL> for block , ( leftpad , rightpad ) in zip ( processed_blocks , pads_used ) : <EOL> result += block [ leftpad : len ( block ) - rightpad ] <EOL> return result <EOL> def _get_split_idx ( N , blocksize , pad = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> num_fullsplits = N // blocksize <EOL> remainder = N % blocksize <EOL> split_idx = [ ] <EOL> pads_used = [ ] <EOL> for i in range ( num_fullsplits ) : <EOL> start = max ( <NUM_LIT:0> , i * blocksize - pad ) <EOL> end = min ( N , ( i + <NUM_LIT:1> ) * blocksize + pad ) <EOL> split_idx . append ( ( start , end ) ) <EOL> leftpad = i * blocksize - start <EOL> rightpad = end - ( i + <NUM_LIT:1> ) * blocksize <EOL> pads_used . append ( ( leftpad , rightpad ) ) <EOL> if remainder : <EOL> start = max ( <NUM_LIT:0> , num_fullsplits * blocksize - pad ) <EOL> split_idx . append ( ( start , N ) ) <EOL> leftpad = num_fullsplits * blocksize - start <EOL> pads_used . append ( ( leftpad , <NUM_LIT:0> ) ) <EOL> return split_idx , pads_used <EOL> def _n_jobs_wrap ( n_jobs ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( n_jobs , int ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % type ( n_jobs ) ) <EOL> if ( n_jobs == <NUM_LIT:0> ) or ( n_jobs < - <NUM_LIT:1> * cpu_count ( ) ) : <EOL> msg = "<STR_LIT>" <EOL> raise ValueError ( "<STR_LIT>" % ( n_jobs , msg ) ) <EOL> if n_jobs < <NUM_LIT:0> : <EOL> n_jobs = max ( cpu_count ( ) + <NUM_LIT:1> + n_jobs , <NUM_LIT:1> ) <EOL> return n_jobs <EOL> def _imap_wrap ( func ) : <EOL> """<STR_LIT>""" <EOL> def wrap ( self , timeout = None ) : <EOL> return func ( self , timeout = timeout if timeout is not None else GOOGLE ) <EOL> return wrap <EOL> def _trypickle ( func ) : <EOL> """<STR_LIT>""" <EOL> genericmsg = "<STR_LIT>" <EOL> boundmethodmsg = genericmsg + '<STR_LIT>' + """<STR_LIT>""" <EOL> try : <EOL> cPickle . dumps ( func ) <EOL> except TypeError as e : <EOL> if '<STR_LIT>' in e . message : <EOL> sys . stderr . write ( boundmethodmsg + "<STR_LIT:\n>" ) <EOL> else : <EOL> sys . stderr . write ( genericmsg + '<STR_LIT:\n>' ) <EOL> raise <EOL> except : <EOL> sys . stderr . write ( genericmsg + '<STR_LIT:\n>' ) <EOL> raise <EOL> IMapUnorderedIterator . next = _imap_wrap ( IMapUnorderedIterator . next ) <EOL> IMapIterator . next = _imap_wrap ( IMapIterator . next ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> pass </s>
