<s> """<STR_LIT>""" <EOL> import abc <EOL> from collections import deque <EOL> from concurrent import futures <EOL> import ctypes <EOL> import functools <EOL> import logging <EOL> import os <EOL> import Queue <EOL> import threading <EOL> import time <EOL> import traceback <EOL> from perfkitbenchmarker import context <EOL> from perfkitbenchmarker import errors <EOL> from perfkitbenchmarker import log_util <EOL> _LONG_TIMEOUT = <NUM_LIT> <EOL> _WAIT_MIN_RECHECK_DELAY = <NUM_LIT> <EOL> _WAIT_MAX_RECHECK_DELAY = <NUM_LIT> <EOL> _THREAD_STOP_PROCESSING = <NUM_LIT:0> <EOL> _THREAD_WAIT_FOR_KEYBOARD_INTERRUPT = <NUM_LIT:1> <EOL> def _GetCallString ( target_arg_tuple ) : <EOL> """<STR_LIT>""" <EOL> target , args , kwargs = target_arg_tuple <EOL> while isinstance ( target , functools . partial ) : <EOL> args = target . args + args <EOL> inner_kwargs = target . keywords . copy ( ) <EOL> inner_kwargs . update ( kwargs ) <EOL> kwargs = inner_kwargs <EOL> target = target . func <EOL> arg_strings = [ str ( a ) for a in args ] <EOL> arg_strings . extend ( [ '<STR_LIT>' . format ( k , v ) for k , v in kwargs . iteritems ( ) ] ) <EOL> return '<STR_LIT>' . format ( getattr ( target , '<STR_LIT>' , target ) , <EOL> '<STR_LIT:U+002CU+0020>' . join ( arg_strings ) ) <EOL> def _WaitForCondition ( condition_callback , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> deadline = None if timeout is None else time . time ( ) + timeout <EOL> delay = _WAIT_MIN_RECHECK_DELAY <EOL> while True : <EOL> if condition_callback ( ) : <EOL> return True <EOL> remaining_time = ( _WAIT_MAX_RECHECK_DELAY if deadline is None <EOL> else deadline - time . time ( ) ) <EOL> if remaining_time <= <NUM_LIT:0> : <EOL> return False <EOL> time . sleep ( delay ) <EOL> delay = min ( delay * <NUM_LIT:2> , remaining_time , _WAIT_MAX_RECHECK_DELAY ) <EOL> class _SingleReaderQueue ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _deque = deque ( ) <EOL> def Get ( self , timeout = None ) : <EOL> if not _WaitForCondition ( lambda : self . _deque , timeout ) : <EOL> raise Queue . Empty <EOL> return self . _deque . popleft ( ) <EOL> def Put ( self , item ) : <EOL> self . _deque . append ( item ) <EOL> class _NonPollingSingleReaderQueue ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _deque = deque ( ) <EOL> self . _lock = threading . Lock ( ) <EOL> self . _lock . acquire ( ) <EOL> def _WaitForItem ( self ) : <EOL> self . _lock . acquire ( ) <EOL> def _SignalAvailableItem ( self ) : <EOL> try : <EOL> self . _lock . release ( ) <EOL> except threading . ThreadError : <EOL> pass <EOL> def Get ( self ) : <EOL> while True : <EOL> self . _WaitForItem ( ) <EOL> if self . _deque : <EOL> item = self . _deque . popleft ( ) <EOL> if self . _deque : <EOL> self . _SignalAvailableItem ( ) <EOL> return item <EOL> def Put ( self , item ) : <EOL> self . _deque . append ( item ) <EOL> self . _SignalAvailableItem ( ) <EOL> class _BackgroundTaskThreadContext ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . benchmark_spec = context . GetThreadBenchmarkSpec ( ) <EOL> self . log_context = log_util . GetThreadLogContext ( ) <EOL> def CopyToCurrentThread ( self ) : <EOL> """<STR_LIT>""" <EOL> log_util . SetThreadLogContext ( log_util . ThreadLogContext ( self . log_context ) ) <EOL> context . SetThreadBenchmarkSpec ( self . benchmark_spec ) <EOL> class _BackgroundTask ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , target , args , kwargs , thread_context ) : <EOL> self . target = target <EOL> self . args = args <EOL> self . kwargs = kwargs <EOL> self . context = thread_context <EOL> self . return_value = None <EOL> self . traceback = None <EOL> def Run ( self ) : <EOL> """<STR_LIT>""" <EOL> self . context . CopyToCurrentThread ( ) <EOL> try : <EOL> self . return_value = self . target ( * self . args , ** self . kwargs ) <EOL> except Exception : <EOL> self . traceback = traceback . format_exc ( ) <EOL> class _BackgroundTaskManager ( object ) : <EOL> """<STR_LIT>""" <EOL> __metaclass__ = abc . ABCMeta <EOL> def __init__ ( self , max_concurrency ) : <EOL> self . _max_concurrency = max_concurrency <EOL> self . tasks = [ ] <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * unused_args , ** unused_kwargs ) : <EOL> pass <EOL> @ abc . abstractmethod <EOL> def StartTask ( self , target , args , kwargs , thread_context ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplemented ( ) <EOL> @ abc . abstractmethod <EOL> def AwaitAnyTask ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplemented ( ) <EOL> @ abc . abstractmethod <EOL> def HandleKeyboardInterrupt ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplemented ( ) <EOL> def _ExecuteBackgroundThreadTasks ( worker_id , task_queue , response_queue ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> response_queue . Put ( worker_id ) <EOL> while True : <EOL> task_tuple = task_queue . Get ( ) <EOL> if task_tuple == _THREAD_STOP_PROCESSING : <EOL> break <EOL> elif task_tuple == _THREAD_WAIT_FOR_KEYBOARD_INTERRUPT : <EOL> while True : <EOL> time . sleep ( _WAIT_MAX_RECHECK_DELAY ) <EOL> task_id , task = task_tuple <EOL> task . Run ( ) <EOL> response_queue . Put ( ( worker_id , task_id ) ) <EOL> except KeyboardInterrupt : <EOL> logging . debug ( '<STR_LIT>' <EOL> '<STR_LIT>' , worker_id , exc_info = True ) <EOL> class _BackgroundThreadTaskManager ( _BackgroundTaskManager ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( _BackgroundThreadTaskManager , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _response_queue = _SingleReaderQueue ( ) <EOL> self . _task_queues = [ ] <EOL> self . _threads = [ ] <EOL> self . _available_worker_ids = range ( self . _max_concurrency ) <EOL> uninitialized_worker_ids = set ( self . _available_worker_ids ) <EOL> for worker_id in self . _available_worker_ids : <EOL> task_queue = _NonPollingSingleReaderQueue ( ) <EOL> self . _task_queues . append ( task_queue ) <EOL> thread = threading . Thread ( <EOL> target = _ExecuteBackgroundThreadTasks , <EOL> args = ( worker_id , task_queue , self . _response_queue ) ) <EOL> thread . daemon = True <EOL> self . _threads . append ( thread ) <EOL> thread . start ( ) <EOL> for _ in self . _threads : <EOL> worker_id = self . _response_queue . Get ( ) <EOL> uninitialized_worker_ids . remove ( worker_id ) <EOL> assert not uninitialized_worker_ids , uninitialized_worker_ids <EOL> def __exit__ ( self , * unused_args , ** unused_kwargs ) : <EOL> for task_queue in self . _task_queues : <EOL> task_queue . Put ( _THREAD_STOP_PROCESSING ) <EOL> for thread in self . _threads : <EOL> _WaitForCondition ( lambda : not thread . is_alive ( ) ) <EOL> def StartTask ( self , target , args , kwargs , thread_context ) : <EOL> assert self . _available_worker_ids , ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> task = _BackgroundTask ( target , args , kwargs , thread_context ) <EOL> task_id = len ( self . tasks ) <EOL> self . tasks . append ( task ) <EOL> worker_id = self . _available_worker_ids . pop ( ) <EOL> self . _task_queues [ worker_id ] . Put ( ( task_id , task ) ) <EOL> def AwaitAnyTask ( self ) : <EOL> worker_id , task_id = self . _response_queue . Get ( ) <EOL> self . _available_worker_ids . append ( worker_id ) <EOL> return task_id <EOL> def HandleKeyboardInterrupt ( self ) : <EOL> for thread in self . _threads : <EOL> ctypes . pythonapi . PyThreadState_SetAsyncExc ( <EOL> ctypes . c_long ( thread . ident ) , ctypes . py_object ( KeyboardInterrupt ) ) <EOL> for task_queue , thread in zip ( self . _task_queues , self . _threads ) : <EOL> task_queue . Put ( _THREAD_WAIT_FOR_KEYBOARD_INTERRUPT ) <EOL> for thread in self . _threads : <EOL> _WaitForCondition ( lambda : not thread . is_alive ( ) ) <EOL> def _ExecuteProcessTask ( task ) : <EOL> """<STR_LIT>""" <EOL> task . Run ( ) <EOL> return task . return_value , task . traceback <EOL> class _BackgroundProcessTaskManager ( _BackgroundTaskManager ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( _BackgroundProcessTaskManager , self ) . __init__ ( * args , ** kwargs ) <EOL> self . _active_futures = { } <EOL> self . _executor = futures . ProcessPoolExecutor ( self . _max_concurrency ) <EOL> def __enter__ ( self ) : <EOL> self . _executor . __enter__ ( ) <EOL> return self <EOL> def __exit__ ( self , * args , ** kwargs ) : <EOL> return self . _executor . __exit__ ( * args , ** kwargs ) <EOL> def StartTask ( self , target , args , kwargs , thread_context ) : <EOL> task = _BackgroundTask ( target , args , kwargs , thread_context ) <EOL> task_id = len ( self . tasks ) <EOL> self . tasks . append ( task ) <EOL> future = self . _executor . submit ( _ExecuteProcessTask , task ) <EOL> self . _active_futures [ future ] = task_id <EOL> def AwaitAnyTask ( self ) : <EOL> completed_tasks = None <EOL> while not completed_tasks : <EOL> completed_tasks , _ = futures . wait ( <EOL> self . _active_futures , timeout = _LONG_TIMEOUT , <EOL> return_when = futures . FIRST_COMPLETED ) <EOL> future = completed_tasks . pop ( ) <EOL> task_id = self . _active_futures . pop ( future ) <EOL> task = self . tasks [ task_id ] <EOL> task . return_value , task . traceback = future . result ( ) <EOL> return task_id <EOL> def HandleKeyboardInterrupt ( self ) : <EOL> self . _executor . shutdown ( wait = True ) <EOL> def _RunParallelTasks ( target_arg_tuples , max_concurrency , get_task_manager , <EOL> parallel_exception_class ) : <EOL> """<STR_LIT>""" <EOL> thread_context = _BackgroundTaskThreadContext ( ) <EOL> max_concurrency = min ( max_concurrency , len ( target_arg_tuples ) ) <EOL> error_strings = [ ] <EOL> started_task_count = <NUM_LIT:0> <EOL> active_task_count = <NUM_LIT:0> <EOL> with get_task_manager ( max_concurrency ) as task_manager : <EOL> try : <EOL> while started_task_count < len ( target_arg_tuples ) or active_task_count : <EOL> if ( started_task_count < len ( target_arg_tuples ) and <EOL> active_task_count < max_concurrency ) : <EOL> target , args , kwargs = target_arg_tuples [ started_task_count ] <EOL> task_manager . StartTask ( target , args , kwargs , thread_context ) <EOL> started_task_count += <NUM_LIT:1> <EOL> active_task_count += <NUM_LIT:1> <EOL> continue <EOL> task_id = task_manager . AwaitAnyTask ( ) <EOL> active_task_count -= <NUM_LIT:1> <EOL> stacktrace = task_manager . tasks [ task_id ] . traceback <EOL> if stacktrace : <EOL> msg = ( '<STR_LIT>' . format ( <EOL> _GetCallString ( target_arg_tuples [ task_id ] ) , os . linesep , <EOL> stacktrace ) ) <EOL> logging . error ( msg ) <EOL> error_strings . append ( msg ) <EOL> except KeyboardInterrupt : <EOL> logging . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , active_task_count ) <EOL> task_manager . HandleKeyboardInterrupt ( ) <EOL> raise <EOL> if error_strings : <EOL> raise parallel_exception_class ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( os . linesep , os . linesep . join ( error_strings ) ) ) <EOL> results = [ task . return_value for task in task_manager . tasks ] <EOL> assert len ( target_arg_tuples ) == len ( results ) , ( target_arg_tuples , results ) <EOL> return results <EOL> def RunParallelThreads ( target_arg_tuples , max_concurrency ) : <EOL> """<STR_LIT>""" <EOL> return _RunParallelTasks ( <EOL> target_arg_tuples , max_concurrency , _BackgroundThreadTaskManager , <EOL> errors . VmUtil . ThreadException ) <EOL> def RunThreaded ( target , thread_params , max_concurrent_threads = <NUM_LIT:200> ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( thread_params , list ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if not thread_params : <EOL> return [ ] <EOL> if not isinstance ( thread_params [ <NUM_LIT:0> ] , tuple ) : <EOL> target_arg_tuples = [ ( target , ( arg , ) , { } ) for arg in thread_params ] <EOL> elif ( not isinstance ( thread_params [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] , tuple ) or <EOL> not isinstance ( thread_params [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] , dict ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> else : <EOL> target_arg_tuples = [ ( target , args , kwargs ) <EOL> for args , kwargs in thread_params ] <EOL> return RunParallelThreads ( target_arg_tuples , <EOL> max_concurrency = max_concurrent_threads ) <EOL> def RunParallelProcesses ( target_arg_tuples , max_concurrency ) : <EOL> """<STR_LIT>""" <EOL> return _RunParallelTasks ( <EOL> target_arg_tuples , max_concurrency , _BackgroundProcessTaskManager , <EOL> errors . VmUtil . CalledProcessException ) </s>
