<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from __future__ import division <EOL> from __future__ import print_function <EOL> import multiprocessing <EOL> import multiprocessing . queues <EOL> import os <EOL> import phlsys_timer <EOL> class MultiprocessingWorkerFinishError ( Exception ) : <EOL> pass <EOL> class CyclingPool ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , job_list , max_workers , max_overrunnable ) : <EOL> """<STR_LIT>""" <EOL> super ( CyclingPool , self ) . __init__ ( ) <EOL> if max_workers < <NUM_LIT:1> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( max_workers ) ) <EOL> if max_overrunnable < <NUM_LIT:0> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( <EOL> max_overrunnable ) ) <EOL> if max_overrunnable >= max_workers : <EOL> raise ValueError ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( <EOL> max_overrunnable , max_workers ) ) <EOL> self . _job_list = job_list <EOL> self . _max_workers = max_workers <EOL> self . _overunnable_workers = _calc_overrunnable_workers ( <EOL> max_workers = max_workers , <EOL> max_overrunnable = max_overrunnable , <EOL> num_jobs = len ( job_list ) ) <EOL> self . _pool_list = _PoolList ( ) <EOL> self . _active_job_index_set = set ( ) <EOL> def cycle_results ( self , overrun_secs ) : <EOL> """<STR_LIT>""" <EOL> timer = phlsys_timer . Timer ( ) <EOL> timer . start ( ) <EOL> def overrun_condition ( ) : <EOL> return timer . duration >= overrun_secs <EOL> for index , result in self . _cycle_results ( overrun_condition ) : <EOL> yield index , result <EOL> def finish_results ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . _pool_list . is_yield_finished ( ) : <EOL> for index , result in self . _overrun_cycle_results ( ) : <EOL> yield index , result <EOL> @ property <EOL> def num_active_jobs ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . _active_job_index_set ) <EOL> def _overrun_cycle_results ( self ) : <EOL> for index , result in self . _pool_list . yield_available_results ( ) : <EOL> self . _active_job_index_set . remove ( index ) <EOL> yield index , result <EOL> def _cycle_results ( self , overrun_condition ) : <EOL> for i , res in self . _overrun_cycle_results ( ) : <EOL> yield i , res <EOL> self . _start_new_cycle ( ) <EOL> should_break = False <EOL> while not should_break : <EOL> should_break = _calc_should_overrun ( <EOL> num_active = self . _pool_list . count_active_workers ( ) , <EOL> num_overrunnable = self . _overunnable_workers , <EOL> condition = overrun_condition , <EOL> is_finished = self . _pool_list . is_yield_finished ( ) ) <EOL> for index , result in self . _overrun_cycle_results ( ) : <EOL> yield index , result <EOL> def _start_new_cycle ( self ) : <EOL> active_workers = self . _pool_list . count_active_workers ( ) <EOL> max_new_workers = self . _max_workers - active_workers <EOL> pool = _Pool ( self . _job_list , max_new_workers ) <EOL> all_job_index_set = set ( xrange ( len ( self . _job_list ) ) ) <EOL> inactive_job_index_set = all_job_index_set - self . _active_job_index_set <EOL> for i in inactive_job_index_set : <EOL> pool . add_job_index ( i ) <EOL> self . _active_job_index_set . add ( i ) <EOL> pool . finish ( ) <EOL> self . _pool_list . add_pool ( pool ) <EOL> def _calc_overrunnable_workers ( max_workers , max_overrunnable , num_jobs ) : <EOL> return min ( <EOL> max_workers - <NUM_LIT:1> , <EOL> num_jobs - <NUM_LIT:1> , <EOL> max_overrunnable ) <EOL> def _calc_should_overrun ( num_active , num_overrunnable , condition , is_finished ) : <EOL> too_busy = num_active > num_overrunnable <EOL> can_overrun = not too_busy and condition ( ) <EOL> return is_finished or can_overrun <EOL> class _PoolList ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _pool_list = [ ] <EOL> def add_pool ( self , pool ) : <EOL> self . _pool_list . append ( pool ) <EOL> def count_active_workers ( self ) : <EOL> active_workers = <NUM_LIT:0> <EOL> for pool in self . _pool_list : <EOL> active_workers += pool . count_active_workers ( ) <EOL> return active_workers <EOL> def yield_available_results ( self ) : <EOL> for pool in self . _pool_list : <EOL> pool . join_finished_workers ( ) <EOL> for index , result in pool . yield_available_results ( ) : <EOL> yield index , result <EOL> finished_pools = [ ] <EOL> for pool in self . _pool_list : <EOL> if pool . is_finished ( ) : <EOL> finished_pools . append ( pool ) <EOL> for pool in finished_pools : <EOL> self . _pool_list . remove ( pool ) <EOL> def is_yield_finished ( self ) : <EOL> return not self . _pool_list <EOL> class _Pool ( object ) : <EOL> def __init__ ( self , job_list , max_workers ) : <EOL> super ( _Pool , self ) . __init__ ( ) <EOL> if max_workers < <NUM_LIT:1> : <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( max_workers ) ) <EOL> mp = multiprocessing <EOL> self . _job_index_queue = mp . queues . SimpleQueue ( ) <EOL> self . _results_queue = mp . queues . SimpleQueue ( ) <EOL> self . _worker_list = [ ] <EOL> num_workers = min ( max_workers , len ( job_list ) ) <EOL> for _ in xrange ( num_workers ) : <EOL> worker = _start_worker_process ( <EOL> job_list , self . _job_index_queue , self . _results_queue ) <EOL> self . _worker_list . append ( worker ) <EOL> def add_job_index ( self , job_index ) : <EOL> self . _job_index_queue . put ( job_index ) <EOL> def finish ( self ) : <EOL> for _ in xrange ( len ( self . _worker_list ) ) : <EOL> self . _job_index_queue . put ( None ) <EOL> def join_finished_workers ( self ) : <EOL> finished_workers = [ ] <EOL> for worker in self . _worker_list : <EOL> if not worker . is_alive ( ) : <EOL> worker . join ( ) <EOL> finished_workers . append ( worker ) <EOL> for worker in finished_workers : <EOL> self . _worker_list . remove ( worker ) <EOL> def yield_available_results ( self ) : <EOL> while not self . _results_queue . empty ( ) : <EOL> yield self . _results_queue . get ( ) <EOL> def count_active_workers ( self ) : <EOL> return len ( self . _worker_list ) <EOL> def is_finished ( self ) : <EOL> return not self . _worker_list <EOL> def _start_worker_process ( job_list , work_queue , results_queue ) : <EOL> worker = multiprocessing . Process ( <EOL> target = _worker_process , <EOL> args = ( job_list , work_queue , results_queue ) ) <EOL> pid = os . getpid ( ) <EOL> try : <EOL> worker . start ( ) <EOL> except AttributeError : <EOL> current_pid = os . getpid ( ) <EOL> if pid != current_pid : <EOL> raise MultiprocessingWorkerFinishError ( <EOL> '<STR_LIT>' . format ( <EOL> current_pid ) ) <EOL> return worker <EOL> def _worker_process ( job_list , work_queue , results_queue ) : <EOL> while True : <EOL> job_index = work_queue . get ( ) <EOL> if job_index is None : <EOL> break <EOL> job = job_list [ job_index ] <EOL> results = job ( ) <EOL> results_queue . put ( ( job_index , results ) ) </s>
