signature:
def launch_if_ready ( self , task_id ) :
    """launch _ if _ ready will launch the specified task , if it is ready
 to run ( for example , without dependencies , and in pending state ) .
 This should be called by any piece of the DataFlowKernel that
 thinks a task may have become ready to run .
 It is not an error to call launch _ if _ ready on a task that is not
 ready to run - launch _ if _ ready will not incorrectly launch that
 task .
 launch _ if _ ready is thread safe , so may be called from any thread
 or callback ."""

implementation:
if self . _count_deps ( self . tasks [ task_id ] [ 'depends' ] ) == 0 : # We can now launch * task *
    new_args , kwargs , exceptions = self . sanitize_and_wrap ( task_id , self . tasks [ task_id ] [ 'args' ] , self . tasks [ task_id ] [ 'kwargs' ] )
    self . tasks [ task_id ] [ 'args' ] = new_args
    self . tasks [ task_id ] [ 'kwargs' ] = kwargs
    if not exceptions : # There are no dependency errors
        exec_fu = None
        # Acquire a lock , retest the state , launch
        with self . tasks [ task_id ] [ 'task_launch_lock' ] :
            if self . tasks [ task_id ] [ 'status' ] == States . pending :
                exec_fu = self . launch_task ( task_id , self . tasks [ task_id ] [ 'func' ] , * new_args , ** kwargs )
        if exec_fu :
            try :
                exec_fu . add_done_callback ( partial ( self . handle_exec_update , task_id ) )
            except Exception as e :
                logger . error ( "add_done_callback got an exception {} which will be ignored" . format ( e ) )
            self . tasks [ task_id ] [ 'exec_fu' ] = exec_fu
            try :
                self . tasks [ task_id ] [ 'app_fu' ] . update_parent ( exec_fu )
                self . tasks [ task_id ] [ 'exec_fu' ] = exec_fu
            except AttributeError as e :
                logger . error ( "Task {}: Caught AttributeError at update_parent" . format ( task_id ) )
                raise e
    else :
        logger . info ( "Task {} failed due to dependency failure" . format ( task_id ) )
        # Raise a dependency exception
        self . tasks [ task_id ] [ 'status' ] = States . dep_fail
        if self . monitoring is not None :
            task_log_info = self . _create_task_log_info ( task_id , 'lazy' )
            self . monitoring . send ( MessageType . TASK_INFO , task_log_info )
        try :
            fu = Future ( )
            fu . retries_left = 0
            self . tasks [ task_id ] [ 'exec_fu' ] = fu
            self . tasks [ task_id ] [ 'app_fu' ] . update_parent ( fu )
            fu . set_exception ( DependencyError ( exceptions , task_id , None ) )
        except AttributeError as e :
            logger . error ( "Task {} AttributeError at update_parent" . format ( task_id ) )
            raise e


signature:
def launch_task ( self , task_id , executable , * args , ** kwargs ) :
    """Handle the actual submission of the task to the executor layer .
 If the app task has the executors attributes not set ( default = = ' all ' )
 the task is launched on a randomly selected executor from the
 list of executors . This behavior could later be updated to support
 binding to executors based on user specified criteria .
 If the app task specifies a particular set of executors , it will be
 targeted at those specific executors .
 Args :
 task _ id ( uuid string ) : A uuid string that uniquely identifies the task
 executable ( callable ) : A callable object
 args ( list of positional args )
 kwargs ( arbitrary keyword arguments )
 Returns :
 Future that tracks the execution of the submitted executable"""

implementation:
self . tasks [ task_id ] [ 'time_submitted' ] = datetime . datetime . now ( )
hit , memo_fu = self . memoizer . check_memo ( task_id , self . tasks [ task_id ] )
if hit :
    logger . info ( "Reusing cached result for task {}" . format ( task_id ) )
    return memo_fu
executor_label = self . tasks [ task_id ] [ "executor" ]
try :
    executor = self . executors [ executor_label ]
except Exception :
    logger . exception ( "Task {} requested invalid executor {}: config is\n{}" . format ( task_id , executor_label , self . _config ) )
if self . monitoring is not None and self . monitoring . resource_monitoring_enabled :
    executable = self . monitoring . monitor_wrapper ( executable , task_id , self . monitoring . monitoring_hub_url , self . run_id , self . monitoring . resource_monitoring_interval )
with self . submitter_lock :
    exec_fu = executor . submit ( executable , * args , ** kwargs )
self . tasks [ task_id ] [ 'status' ] = States . launched
if self . monitoring is not None :
    task_log_info = self . _create_task_log_info ( task_id , 'lazy' )
    self . monitoring . send ( MessageType . TASK_INFO , task_log_info )
exec_fu . retries_left = self . _config . retries - self . tasks [ task_id ] [ 'fail_count' ]
logger . info ( "Task {} launched on executor {}" . format ( task_id , executor . label ) )
return exec_fu


signature:
def _add_input_deps ( self , executor , args , kwargs ) :
    """Look for inputs of the app that are remote files . Submit stage _ in
 apps for such files and replace the file objects in the inputs list with
 corresponding DataFuture objects .
 Args :
 - executor ( str ) : executor where the app is going to be launched
 - args ( List ) : Positional args to app function
 - kwargs ( Dict ) : Kwargs to app function"""

implementation:
# Return if the task is _ * _ stage _ in
if executor == 'data_manager' :
    return args , kwargs
inputs = kwargs . get ( 'inputs' , [ ] )
for idx , f in enumerate ( inputs ) :
    if isinstance ( f , File ) and f . is_remote ( ) :
        inputs [ idx ] = self . data_manager . stage_in ( f , executor )
for kwarg , f in kwargs . items ( ) :
    if isinstance ( f , File ) and f . is_remote ( ) :
        kwargs [ kwarg ] = self . data_manager . stage_in ( f , executor )
newargs = list ( args )
for idx , f in enumerate ( newargs ) :
    if isinstance ( f , File ) and f . is_remote ( ) :
        newargs [ idx ] = self . data_manager . stage_in ( f , executor )
return tuple ( newargs ) , kwargs


signature:
def _gather_all_deps ( self , args , kwargs ) :
    """Count the number of unresolved futures on which a task depends .
 Args :
 - args ( List [ args ] ) : The list of args list to the fn
 - kwargs ( Dict { kwargs } ) : The dict of all kwargs passed to the fn
 Returns :
 - count , [ list of dependencies ]"""

implementation:
# Check the positional args
depends = [ ]
count = 0
for dep in args :
    if isinstance ( dep , Future ) :
        if self . tasks [ dep . tid ] [ 'status' ] not in FINAL_STATES :
            count += 1
        depends . extend ( [ dep ] )
# Check for explicit kwargs ex , fu _ 1 = < fut >
for key in kwargs :
    dep = kwargs [ key ]
    if isinstance ( dep , Future ) :
        if self . tasks [ dep . tid ] [ 'status' ] not in FINAL_STATES :
            count += 1
        depends . extend ( [ dep ] )
# Check for futures in inputs = [ < fut > . . . ]
for dep in kwargs . get ( 'inputs' , [ ] ) :
    if isinstance ( dep , Future ) :
        if self . tasks [ dep . tid ] [ 'status' ] not in FINAL_STATES :
            count += 1
        depends . extend ( [ dep ] )
return count , depends


signature:
def sanitize_and_wrap ( self , task_id , args , kwargs ) :
    """This function should be called * * ONLY * * when all the futures we track have been resolved .
 If the user hid futures a level below , we will not catch
 it , and will ( most likely ) result in a type error .
 Args :
 task _ id ( uuid str ) : Task id
 func ( Function ) : App function
 args ( List ) : Positional args to app function
 kwargs ( Dict ) : Kwargs to app function
 Return :
 partial function evaluated with all dependencies in args , kwargs and kwargs [ ' inputs ' ] evaluated ."""

implementation:
dep_failures = [ ]
# Replace item in args
new_args = [ ]
for dep in args :
    if isinstance ( dep , Future ) :
        try :
            new_args . extend ( [ dep . result ( ) ] )
        except Exception as e :
            if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES :
                dep_failures . extend ( [ e ] )
    else :
        new_args . extend ( [ dep ] )
# Check for explicit kwargs ex , fu _ 1 = < fut >
for key in kwargs :
    dep = kwargs [ key ]
    if isinstance ( dep , Future ) :
        try :
            kwargs [ key ] = dep . result ( )
        except Exception as e :
            if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES :
                dep_failures . extend ( [ e ] )
# Check for futures in inputs = [ < fut > . . . ]
if 'inputs' in kwargs :
    new_inputs = [ ]
    for dep in kwargs [ 'inputs' ] :
        if isinstance ( dep , Future ) :
            try :
                new_inputs . extend ( [ dep . result ( ) ] )
            except Exception as e :
                if self . tasks [ dep . tid ] [ 'status' ] in FINAL_FAILURE_STATES :
                    dep_failures . extend ( [ e ] )
        else :
            new_inputs . extend ( [ dep ] )
    kwargs [ 'inputs' ] = new_inputs
return new_args , kwargs , dep_failures


signature:
def submit ( self , func , * args , executors = 'all' , fn_hash = None , cache = False , ** kwargs ) :
    """Add task to the dataflow system .
 If the app task has the executors attributes not set ( default = = ' all ' )
 the task will be launched on a randomly selected executor from the
 list of executors . If the app task specifies a particular set of
 executors , it will be targeted at the specified executors .
 > > > IF all deps are met :
 > > > send to the runnable queue and launch the task
 > > > ELSE :
 > > > post the task in the pending queue
 Args :
 - func : A function object
 - * args : Args to the function
 KWargs :
 - executors ( list or string ) : List of executors this call could go to .
 Default = ' all '
 - fn _ hash ( Str ) : Hash of the function and inputs
 Default = None
 - cache ( Bool ) : To enable memoization or not
 - kwargs ( dict ) : Rest of the kwargs to the fn passed as dict .
 Returns :
 ( AppFuture ) [ DataFutures , ]"""

implementation:
if self . cleanup_called :
    raise ValueError ( "Cannot submit to a DFK that has been cleaned up" )
task_id = self . task_count
self . task_count += 1
if isinstance ( executors , str ) and executors . lower ( ) == 'all' :
    choices = list ( e for e in self . executors if e != 'data_manager' )
elif isinstance ( executors , list ) :
    choices = executors
executor = random . choice ( choices )
# Transform remote input files to data futures
args , kwargs = self . _add_input_deps ( executor , args , kwargs )
task_def = { 'depends' : None , 'executor' : executor , 'func' : func , 'func_name' : func . __name__ , 'args' : args , 'kwargs' : kwargs , 'fn_hash' : fn_hash , 'memoize' : cache , 'callback' : None , 'exec_fu' : None , 'checkpoint' : None , 'fail_count' : 0 , 'fail_history' : [ ] , 'env' : None , 'status' : States . unsched , 'id' : task_id , 'time_submitted' : None , 'time_returned' : None , 'app_fu' : None }
if task_id in self . tasks :
    raise DuplicateTaskError ( "internal consistency error: Task {0} already exists in task list" . format ( task_id ) )
else :
    self . tasks [ task_id ] = task_def
# Get the dep count and a list of dependencies for the task
dep_cnt , depends = self . _gather_all_deps ( args , kwargs )
self . tasks [ task_id ] [ 'depends' ] = depends
# Extract stdout and stderr to pass to AppFuture :
task_stdout = kwargs . get ( 'stdout' )
task_stderr = kwargs . get ( 'stderr' )
logger . info ( "Task {} submitted for App {}, waiting on tasks {}" . format ( task_id , task_def [ 'func_name' ] , [ fu . tid for fu in depends ] ) )
self . tasks [ task_id ] [ 'task_launch_lock' ] = threading . Lock ( )
app_fu = AppFuture ( tid = task_id , stdout = task_stdout , stderr = task_stderr )
self . tasks [ task_id ] [ 'app_fu' ] = app_fu
app_fu . add_done_callback ( partial ( self . handle_app_update , task_id ) )
self . tasks [ task_id ] [ 'status' ] = States . pending
logger . debug ( "Task {} set to pending state with AppFuture: {}" . format ( task_id , task_def [ 'app_fu' ] ) )
# at this point add callbacks to all dependencies to do a launch _ if _ ready
# call whenever a dependency completes .
# we need to be careful about the order of setting the state to pending ,
# adding the callbacks , and caling launch _ if _ ready explicitly once always below .
# I think as long as we call launch _ if _ ready once after setting pending , then
# we can add the callback dependencies at any point : if the callbacks all fire
# before then , they won ' t cause a launch , but the one below will . if they fire
# after we set it pending , then the last one will cause a launch , and the
# explicit one won ' t .
for d in depends :
    def callback_adapter ( dep_fut ) :
        self . launch_if_ready ( task_id )
    try :
        d . add_done_callback ( callback_adapter )
    except Exception as e :
        logger . error ( "add_done_callback got an exception {} which will be ignored" . format ( e ) )
self . launch_if_ready ( task_id )
return task_def [ 'app_fu' ]


signature:
def wait_for_current_tasks ( self ) :
    """Waits for all tasks in the task list to be completed , by waiting for their
 AppFuture to be completed . This method will not necessarily wait for any tasks
 added after cleanup has started ( such as data stageout ? )"""

implementation:
logger . info ( "Waiting for all remaining tasks to complete" )
for task_id in self . tasks : # . exception ( ) is a less exception throwing way of
# waiting for completion than . result ( )
    fut = self . tasks [ task_id ] [ 'app_fu' ]
    if not fut . done ( ) :
        logger . debug ( "Waiting for task {} to complete" . format ( task_id ) )
        fut . exception ( )
logger . info ( "All remaining tasks completed" )


signature:
def checkpoint ( self , tasks = None ) :
    """Checkpoint the dfk incrementally to a checkpoint file .
 When called , every task that has been completed yet not
 checkpointed is checkpointed to a file .
 Kwargs :
 - tasks ( List of task ids ) : List of task ids to checkpoint . Default = None
 if set to None , we iterate over all tasks held by the DFK .
 . . note : :
 Checkpointing only works if memoization is enabled
 Returns :
 Checkpoint dir if checkpoints were written successfully .
 By default the checkpoints are written to the RUNDIR of the current
 run under RUNDIR / checkpoints / { tasks . pkl , dfk . pkl }"""

implementation:
with self . checkpoint_lock :
    checkpoint_queue = None
    if tasks :
        checkpoint_queue = tasks
    else :
        checkpoint_queue = self . tasks
    checkpoint_dir = '{0}/checkpoint' . format ( self . run_dir )
    checkpoint_dfk = checkpoint_dir + '/dfk.pkl'
    checkpoint_tasks = checkpoint_dir + '/tasks.pkl'
    if not os . path . exists ( checkpoint_dir ) :
        try :
            os . makedirs ( checkpoint_dir )
        except FileExistsError :
            pass
    with open ( checkpoint_dfk , 'wb' ) as f :
        state = { 'rundir' : self . run_dir , 'task_count' : self . task_count }
        pickle . dump ( state , f )
    count = 0
    with open ( checkpoint_tasks , 'ab' ) as f :
        for task_id in checkpoint_queue :
            if not self . tasks [ task_id ] [ 'checkpoint' ] and self . tasks [ task_id ] [ 'app_fu' ] . done ( ) and self . tasks [ task_id ] [ 'app_fu' ] . exception ( ) is None :
                hashsum = self . tasks [ task_id ] [ 'hashsum' ]
                if not hashsum :
                    continue
                t = { 'hash' : hashsum , 'exception' : None , 'result' : None }
                try : # Asking for the result will raise an exception if
                # the app had failed . Should we even checkpoint these ?
                # TODO : Resolve this question ?
                    r = self . memoizer . hash_lookup ( hashsum ) . result ( )
                except Exception as e :
                    t [ 'exception' ] = e
                else :
                    t [ 'result' ] = r
                # We are using pickle here since pickle dumps to a file in ' ab '
                # mode behave like a incremental log .
                pickle . dump ( t , f )
                count += 1
                self . tasks [ task_id ] [ 'checkpoint' ] = True
                logger . debug ( "Task {} checkpointed" . format ( task_id ) )
    self . checkpointed_tasks += count
    if count == 0 :
        if self . checkpointed_tasks == 0 :
            logger . warn ( "No tasks checkpointed so far in this run. Please ensure caching is enabled" )
        else :
            logger . debug ( "No tasks checkpointed in this pass." )
    else :
        logger . info ( "Done checkpointing {} tasks" . format ( count ) )
    return checkpoint_dir


signature:
def _load_checkpoints ( self , checkpointDirs ) :
    """Load a checkpoint file into a lookup table .
 The data being loaded from the pickle file mostly contains input
 attributes of the task : func , args , kwargs , env . . .
 To simplify the check of whether the exact task has been completed
 in the checkpoint , we hash these input params and use it as the key
 for the memoized lookup table .
 Args :
 - checkpointDirs ( list ) : List of filepaths to checkpoints
 Eg . [ ' runinfo / 001 ' , ' runinfo / 002 ' ]
 Returns :
 - memoized _ lookup _ table ( dict )"""

implementation:
memo_lookup_table = { }
for checkpoint_dir in checkpointDirs :
    logger . info ( "Loading checkpoints from {}" . format ( checkpoint_dir ) )
    checkpoint_file = os . path . join ( checkpoint_dir , 'tasks.pkl' )
    try :
        with open ( checkpoint_file , 'rb' ) as f :
            while True :
                try :
                    data = pickle . load ( f )
                    # Copy and hash only the input attributes
                    memo_fu = Future ( )
                    if data [ 'exception' ] :
                        memo_fu . set_exception ( data [ 'exception' ] )
                    else :
                        memo_fu . set_result ( data [ 'result' ] )
                    memo_lookup_table [ data [ 'hash' ] ] = memo_fu
                except EOFError : # Done with the checkpoint file
                    break
    except FileNotFoundError :
        reason = "Checkpoint file was not found: {}" . format ( checkpoint_file )
        logger . error ( reason )
        raise BadCheckpoint ( reason )
    except Exception :
        reason = "Failed to load checkpoint: {}" . format ( checkpoint_file )
        logger . error ( reason )
        raise BadCheckpoint ( reason )
    logger . info ( "Completed loading checkpoint:{0} with {1} tasks" . format ( checkpoint_file , len ( memo_lookup_table . keys ( ) ) ) )
return memo_lookup_table


signature:
def load_checkpoints ( self , checkpointDirs ) :
    """Load checkpoints from the checkpoint files into a dictionary .
 The results are used to pre - populate the memoizer ' s lookup _ table
 Kwargs :
 - checkpointDirs ( list ) : List of run folder to use as checkpoints
 Eg . [ ' runinfo / 001 ' , ' runinfo / 002 ' ]
 Returns :
 - dict containing , hashed - > future mappings"""

implementation:
self . memo_lookup_table = None
if not checkpointDirs :
    return { }
if type ( checkpointDirs ) is not list :
    raise BadCheckpoint ( "checkpointDirs expects a list of checkpoints" )
return self . _load_checkpoints ( checkpointDirs )


signature:
def load ( cls , config : Optional [ Config ] = None ) :
    """Load a DataFlowKernel .
 Args :
 - config ( Config ) : Configuration to load . This config will be passed to a
 new DataFlowKernel instantiation which will be set as the active DataFlowKernel .
 Returns :
 - DataFlowKernel : The loaded DataFlowKernel object ."""

implementation:
if cls . _dfk is not None :
    raise RuntimeError ( 'Config has already been loaded' )
if config is None :
    cls . _dfk = DataFlowKernel ( Config ( ) )
else :
    cls . _dfk = DataFlowKernel ( config )
return cls . _dfk


signature:
def starter ( comm_q , * args , ** kwargs ) :
    """Start the interchange process
 The executor is expected to call this function . The args , kwargs match that of the Interchange . _ _ init _ _"""

implementation:
# logger = multiprocessing . get _ logger ( )
ic = Interchange ( * args , ** kwargs )
comm_q . put ( ( ic . worker_task_port , ic . worker_result_port ) )
ic . start ( )


signature:
def get_tasks ( self , count ) :
    """Obtains a batch of tasks from the internal pending _ task _ queue
 Parameters
 count : int
 Count of tasks to get from the queue
 Returns
 List of upto count tasks . May return fewer than count down to an empty list
 eg . [ { ' task _ id ' : < x > , ' buffer ' : < buf > } . . . ]"""

implementation:
tasks = [ ]
for i in range ( 0 , count ) :
    try :
        x = self . pending_task_queue . get ( block = False )
    except queue . Empty :
        break
    else :
        tasks . append ( x )
return tasks


signature:
def migrate_tasks_to_internal ( self , kill_event ) :
    """Pull tasks from the incoming tasks 0mq pipe onto the internal
 pending task queue
 Parameters :
 kill _ event : threading . Event
 Event to let the thread know when it is time to die ."""

implementation:
logger . info ( "[TASK_PULL_THREAD] Starting" )
task_counter = 0
poller = zmq . Poller ( )
poller . register ( self . task_incoming , zmq . POLLIN )
while not kill_event . is_set ( ) :
    try :
        msg = self . task_incoming . recv_pyobj ( )
    except zmq . Again : # We just timed out while attempting to receive
        logger . debug ( "[TASK_PULL_THREAD] {} tasks in internal queue" . format ( self . pending_task_queue . qsize ( ) ) )
        continue
    if msg == 'STOP' :
        kill_event . set ( )
        break
    else :
        self . pending_task_queue . put ( msg )
        task_counter += 1
        logger . debug ( "[TASK_PULL_THREAD] Fetched task:{}" . format ( task_counter ) )


signature:
def _command_server ( self , kill_event ) :
    """Command server to run async command to the interchange"""

implementation:
logger . debug ( "[COMMAND] Command Server Starting" )
while not kill_event . is_set ( ) :
    try :
        command_req = self . command_channel . recv_pyobj ( )
        logger . debug ( "[COMMAND] Received command request: {}" . format ( command_req ) )
        if command_req == "OUTSTANDING_C" :
            outstanding = self . pending_task_queue . qsize ( )
            for manager in self . _ready_manager_queue :
                outstanding += len ( self . _ready_manager_queue [ manager ] [ 'tasks' ] )
            reply = outstanding
        elif command_req == "WORKERS" :
            num_workers = 0
            for manager in self . _ready_manager_queue :
                num_workers += self . _ready_manager_queue [ manager ] [ 'worker_count' ]
            reply = num_workers
        elif command_req == "MANAGERS" :
            reply = [ ]
            for manager in self . _ready_manager_queue :
                resp = { 'manager' : manager . decode ( 'utf-8' ) , 'block_id' : self . _ready_manager_queue [ manager ] [ 'block_id' ] , 'worker_count' : self . _ready_manager_queue [ manager ] [ 'worker_count' ] , 'tasks' : len ( self . _ready_manager_queue [ manager ] [ 'tasks' ] ) , 'active' : self . _ready_manager_queue [ manager ] [ 'active' ] }
                reply . append ( resp )
        elif command_req . startswith ( "HOLD_WORKER" ) :
            cmd , s_manager = command_req . split ( ';' )
            manager = s_manager . encode ( 'utf-8' )
            logger . info ( "[CMD] Received HOLD_WORKER for {}" . format ( manager ) )
            if manager in self . _ready_manager_queue :
                self . _ready_manager_queue [ manager ] [ 'active' ] = False
                reply = True
            else :
                reply = False
        elif command_req == "SHUTDOWN" :
            logger . info ( "[CMD] Received SHUTDOWN command" )
            kill_event . set ( )
            reply = True
        else :
            reply = None
        logger . debug ( "[COMMAND] Reply: {}" . format ( reply ) )
        self . command_channel . send_pyobj ( reply )
    except zmq . Again :
        logger . debug ( "[COMMAND] is alive" )
        continue


signature:
def start ( self , poll_period = None ) :
    """Start the NeedNameQeueu
 Parameters :
 TODO : Move task receiving to a thread"""

implementation:
logger . info ( "Incoming ports bound" )
if poll_period is None :
    poll_period = self . poll_period
start = time . time ( )
count = 0
self . _kill_event = threading . Event ( )
self . _task_puller_thread = threading . Thread ( target = self . migrate_tasks_to_internal , args = ( self . _kill_event , ) )
self . _task_puller_thread . start ( )
self . _command_thread = threading . Thread ( target = self . _command_server , args = ( self . _kill_event , ) )
self . _command_thread . start ( )
poller = zmq . Poller ( )
# poller . register ( self . task _ incoming , zmq . POLLIN )
poller . register ( self . task_outgoing , zmq . POLLIN )
poller . register ( self . results_incoming , zmq . POLLIN )
# These are managers which we should examine in an iteration
# for scheduling a job ( or maybe any other attention ? ) .
# Anything altering the state of the manager should add it
# onto this list .
interesting_managers = set ( )
while not self . _kill_event . is_set ( ) :
    self . socks = dict ( poller . poll ( timeout = poll_period ) )
    # Listen for requests for work
    if self . task_outgoing in self . socks and self . socks [ self . task_outgoing ] == zmq . POLLIN :
        logger . debug ( "[MAIN] starting task_outgoing section" )
        message = self . task_outgoing . recv_multipart ( )
        manager = message [ 0 ]
        if manager not in self . _ready_manager_queue :
            reg_flag = False
            try :
                msg = json . loads ( message [ 1 ] . decode ( 'utf-8' ) )
                reg_flag = True
            except Exception :
                logger . warning ( "[MAIN] Got a non-json registration message from manager:{}" . format ( manager ) )
                logger . debug ( "[MAIN] Message :\n{}\n" . format ( message [ 0 ] ) )
            # By default we set up to ignore bad nodes / registration messages .
            self . _ready_manager_queue [ manager ] = { 'last' : time . time ( ) , 'free_capacity' : 0 , 'block_id' : None , 'max_capacity' : 0 , 'active' : True , 'tasks' : [ ] }
            if reg_flag is True :
                interesting_managers . add ( manager )
                logger . info ( "[MAIN] Adding manager: {} to ready queue" . format ( manager ) )
                self . _ready_manager_queue [ manager ] . update ( msg )
                logger . info ( "[MAIN] Registration info for manager {}: {}" . format ( manager , msg ) )
                if ( msg [ 'python_v' ] . rsplit ( "." , 1 ) [ 0 ] != self . current_platform [ 'python_v' ] . rsplit ( "." , 1 ) [ 0 ] or msg [ 'parsl_v' ] != self . current_platform [ 'parsl_v' ] ) :
                    logger . warn ( "[MAIN] Manager {} has incompatible version info with the interchange" . format ( manager ) )
                    if self . suppress_failure is False :
                        logger . debug ( "Setting kill event" )
                        self . _kill_event . set ( )
                        e = ManagerLost ( manager )
                        result_package = { 'task_id' : - 1 , 'exception' : serialize_object ( e ) }
                        pkl_package = pickle . dumps ( result_package )
                        self . results_outgoing . send ( pkl_package )
                        logger . warning ( "[MAIN] Sent failure reports, unregistering manager" )
                    else :
                        logger . debug ( "[MAIN] Suppressing shutdown due to version incompatibility" )
                else :
                    logger . info ( "[MAIN] Manager {} has compatible Parsl version {}" . format ( manager , msg [ 'parsl_v' ] ) )
                    logger . info ( "[MAIN] Manager {} has compatible Python version {}" . format ( manager , msg [ 'python_v' ] . rsplit ( "." , 1 ) [ 0 ] ) )
            else : # Registration has failed .
                if self . suppress_failure is False :
                    self . _kill_event . set ( )
                    e = BadRegistration ( manager , critical = True )
                    result_package = { 'task_id' : - 1 , 'exception' : serialize_object ( e ) }
                    pkl_package = pickle . dumps ( result_package )
                    self . results_outgoing . send ( pkl_package )
                else :
                    logger . debug ( "[MAIN] Suppressing bad registration from manager:{}" . format ( manager ) )
        else :
            tasks_requested = int . from_bytes ( message [ 1 ] , "little" )
            self . _ready_manager_queue [ manager ] [ 'last' ] = time . time ( )
            if tasks_requested == HEARTBEAT_CODE :
                logger . debug ( "[MAIN] Manager {} sent heartbeat" . format ( manager ) )
                self . task_outgoing . send_multipart ( [ manager , b'' , PKL_HEARTBEAT_CODE ] )
            else :
                logger . debug ( "[MAIN] Manager {} requested {} tasks" . format ( manager , tasks_requested ) )
                self . _ready_manager_queue [ manager ] [ 'free_capacity' ] = tasks_requested
                interesting_managers . add ( manager )
        logger . debug ( "[MAIN] leaving task_outgoing section" )
    # If we had received any requests , check if there are tasks that could be passed
    logger . debug ( "Managers count (total/interesting): {}/{}" . format ( len ( self . _ready_manager_queue ) , len ( interesting_managers ) ) )
    if interesting_managers and not self . pending_task_queue . empty ( ) :
        shuffled_managers = list ( interesting_managers )
        random . shuffle ( shuffled_managers )
        while shuffled_managers and not self . pending_task_queue . empty ( ) : # cf . the if statement above . . .
            manager = shuffled_managers . pop ( )
            tasks_inflight = len ( self . _ready_manager_queue [ manager ] [ 'tasks' ] )
            real_capacity = min ( self . _ready_manager_queue [ manager ] [ 'free_capacity' ] , self . _ready_manager_queue [ manager ] [ 'max_capacity' ] - tasks_inflight )
            if ( real_capacity and self . _ready_manager_queue [ manager ] [ 'active' ] ) :
                tasks = self . get_tasks ( real_capacity )
                if tasks :
                    self . task_outgoing . send_multipart ( [ manager , b'' , pickle . dumps ( tasks ) ] )
                    task_count = len ( tasks )
                    count += task_count
                    tids = [ t [ 'task_id' ] for t in tasks ]
                    self . _ready_manager_queue [ manager ] [ 'free_capacity' ] -= task_count
                    self . _ready_manager_queue [ manager ] [ 'tasks' ] . extend ( tids )
                    logger . debug ( "[MAIN] Sent tasks: {} to manager {}" . format ( tids , manager ) )
                    if self . _ready_manager_queue [ manager ] [ 'free_capacity' ] > 0 :
                        logger . debug ( "[MAIN] Manager {} has free_capacity {}" . format ( manager , self . _ready_manager_queue [ manager ] [ 'free_capacity' ] ) )
                        # . . . so keep it in the interesting _ managers list
                    else :
                        logger . debug ( "[MAIN] Manager {} is now saturated" . format ( manager ) )
                        interesting_managers . remove ( manager )
            else :
                interesting_managers . remove ( manager )
                # logger . debug ( " Nothing to send to manager { } " . format ( manager ) )
        logger . debug ( "[MAIN] leaving _ready_manager_queue section, with {} managers still interesting" . format ( len ( interesting_managers ) ) )
    else :
        logger . debug ( "[MAIN] either no interesting managers or no tasks, so skipping manager pass" )
    # Receive any results and forward to client
    if self . results_incoming in self . socks and self . socks [ self . results_incoming ] == zmq . POLLIN :
        logger . debug ( "[MAIN] entering results_incoming section" )
        manager , * b_messages = self . results_incoming . recv_multipart ( )
        if manager not in self . _ready_manager_queue :
            logger . warning ( "[MAIN] Received a result from a un-registered manager: {}" . format ( manager ) )
        else :
            logger . debug ( "[MAIN] Got {} result items in batch" . format ( len ( b_messages ) ) )
            for b_message in b_messages :
                r = pickle . loads ( b_message )
                # logger . debug ( " [ MAIN ] Received result for task { } from { } " . format ( r [ ' task _ id ' ] , manager ) )
                self . _ready_manager_queue [ manager ] [ 'tasks' ] . remove ( r [ 'task_id' ] )
            self . results_outgoing . send_multipart ( b_messages )
            logger . debug ( "[MAIN] Current tasks: {}" . format ( self . _ready_manager_queue [ manager ] [ 'tasks' ] ) )
        logger . debug ( "[MAIN] leaving results_incoming section" )
    logger . debug ( "[MAIN] entering bad_managers section" )
    bad_managers = [ manager for manager in self . _ready_manager_queue if time . time ( ) - self . _ready_manager_queue [ manager ] [ 'last' ] > self . heartbeat_threshold ]
    for manager in bad_managers :
        logger . debug ( "[MAIN] Last: {} Current: {}" . format ( self . _ready_manager_queue [ manager ] [ 'last' ] , time . time ( ) ) )
        logger . warning ( "[MAIN] Too many heartbeats missed for manager {}" . format ( manager ) )
        for tid in self . _ready_manager_queue [ manager ] [ 'tasks' ] :
            try :
                raise ManagerLost ( manager )
            except Exception :
                result_package = { 'task_id' : tid , 'exception' : serialize_object ( RemoteExceptionWrapper ( * sys . exc_info ( ) ) ) }
                pkl_package = pickle . dumps ( result_package )
                self . results_outgoing . send ( pkl_package )
                logger . warning ( "[MAIN] Sent failure reports, unregistering manager" )
        self . _ready_manager_queue . pop ( manager , 'None' )
    logger . debug ( "[MAIN] leaving bad_managers section" )
    logger . debug ( "[MAIN] ending one main loop iteration" )
delta = time . time ( ) - start
logger . info ( "Processed {} tasks in {} seconds" . format ( count , delta ) )
logger . warning ( "Exiting" )


implementation:
# logger = multiprocessing . get _ logger ( )
ic = Interchange ( * args , ** kwargs )
comm_q . put ( ic . worker_port )
ic . start ( )
logger . debug ( "Port information sent back to client" )


signature:
def start ( self ) :
    """TODO : docstring"""

implementation:
logger . info ( "Starting interchange" )
# last = time . time ( )
while True : # active _ flag = False
    socks = dict ( self . poller . poll ( 1 ) )
    if socks . get ( self . task_incoming ) == zmq . POLLIN :
        message = self . task_incoming . recv_multipart ( )
        logger . debug ( "Got new task from client" )
        self . worker_messages . send_multipart ( message )
        logger . debug ( "Sent task to worker" )
        # active _ flag = True
        # last = time . time ( )
    if socks . get ( self . worker_messages ) == zmq . POLLIN :
        message = self . worker_messages . recv_multipart ( )
        logger . debug ( "Got new result from worker" )
        # self . result _ outgoing . send _ multipart ( message )
        self . result_outgoing . send_multipart ( message [ 1 : ] )
        logger . debug ( "Sent result to client" )


signature:
def execute_task ( bufs ) :
    """Deserialize the buffer and execute the task .
 Returns the result or throws exception ."""

implementation:
user_ns = locals ( )
user_ns . update ( { '__builtins__' : __builtins__ } )
f , args , kwargs = unpack_apply_message ( bufs , user_ns , copy = False )
# We might need to look into callability of the function from itself
# since we change it ' s name in the new namespace
prefix = "parsl_"
fname = prefix + "f"
argname = prefix + "args"
kwargname = prefix + "kwargs"
resultname = prefix + "result"
user_ns . update ( { fname : f , argname : args , kwargname : kwargs , resultname : resultname } )
code = "{0} = {1}(*{2}, **{3})" . format ( resultname , fname , argname , kwargname )
try : # logger . debug ( " [ RUNNER ] Executing : { 0 } " . format ( code ) )
    exec ( code , user_ns , user_ns )
except Exception as e :
    logger . warning ( "Caught exception; will raise it: {}" . format ( e ) , exc_info = True )
    raise e
else : # logger . debug ( " [ RUNNER ] Result : { 0 } " . format ( user _ ns . get ( resultname ) ) )
    return user_ns . get ( resultname )


signature:
def push_results ( self , kill_event ) :
    """Listens on the pending _ result _ queue and sends out results via 0mq
 Parameters :
 kill _ event : threading . Event
 Event to let the thread know when it is time to die ."""

implementation:
logger . debug ( "[RESULT_PUSH_THREAD] Starting thread" )
push_poll_period = max ( 10 , self . poll_period ) / 1000
# push _ poll _ period must be atleast 10 ms
logger . debug ( "[RESULT_PUSH_THREAD] push poll period: {}" . format ( push_poll_period ) )
last_beat = time . time ( )
items = [ ]
while not kill_event . is_set ( ) :
    try :
        r = self . pending_result_queue . get ( block = True , timeout = push_poll_period )
        items . append ( r )
    except queue . Empty :
        pass
    except Exception as e :
        logger . exception ( "[RESULT_PUSH_THREAD] Got an exception: {}" . format ( e ) )
    # If we have reached poll _ period duration or timer has expired , we send results
    if len ( items ) >= self . max_queue_size or time . time ( ) > last_beat + push_poll_period :
        last_beat = time . time ( )
        if items :
            self . result_outgoing . send_multipart ( items )
            items = [ ]
logger . critical ( "[RESULT_PUSH_THREAD] Exiting" )


signature:
def start ( self ) :
    """Start the worker processes .
 TODO : Move task receiving to a thread"""

implementation:
start = time . time ( )
self . _kill_event = threading . Event ( )
self . procs = { }
for worker_id in range ( self . worker_count ) :
    p = multiprocessing . Process ( target = worker , args = ( worker_id , self . uid , self . pending_task_queue , self . pending_result_queue , self . ready_worker_queue , ) )
    p . start ( )
    self . procs [ worker_id ] = p
logger . debug ( "Manager synced with workers" )
self . _task_puller_thread = threading . Thread ( target = self . pull_tasks , args = ( self . _kill_event , ) )
self . _result_pusher_thread = threading . Thread ( target = self . push_results , args = ( self . _kill_event , ) )
self . _task_puller_thread . start ( )
self . _result_pusher_thread . start ( )
logger . info ( "Loop start" )
# TODO : Add mechanism in this loop to stop the worker pool
# This might need a multiprocessing event to signal back .
self . _kill_event . wait ( )
logger . critical ( "[MAIN] Received kill event, terminating worker processes" )
self . _task_puller_thread . join ( )
self . _result_pusher_thread . join ( )
for proc_id in self . procs :
    self . procs [ proc_id ] . terminate ( )
    logger . critical ( "Terminating worker {}:{}" . format ( self . procs [ proc_id ] , self . procs [ proc_id ] . is_alive ( ) ) )
    self . procs [ proc_id ] . join ( )
    logger . debug ( "Worker:{} joined successfully" . format ( self . procs [ proc_id ] ) )
self . task_incoming . close ( )
self . result_outgoing . close ( )
self . context . term ( )
delta = time . time ( ) - start
logger . info ( "process_worker_pool ran for {} seconds" . format ( delta ) )
return


signature:
def get_data_manager ( cls ) :
    """Return the DataManager of the currently loaded DataFlowKernel ."""

implementation:
from parsl . dataflow . dflow import DataFlowKernelLoader
dfk = DataFlowKernelLoader . dfk ( )
return dfk . executors [ 'data_manager' ]


signature:
def shutdown ( self , block = False ) :
    """Shutdown the ThreadPool .
 Kwargs :
 - block ( bool ) : To block for confirmations or not"""

implementation:
x = self . executor . shutdown ( wait = block )
logger . debug ( "Done with executor shutdown" )
return x


signature:
def stage_in ( self , file , executor ) :
    """Transport the file from the input source to the executor .
 This function returns a DataFuture .
 Args :
 - self
 - file ( File ) : file to stage in
 - executor ( str ) : an executor the file is going to be staged in to .
 If the executor argument is not specified for a file
 with ' globus ' scheme , the file will be staged in to
 the first executor with the " globus " key in a config ."""

implementation:
if file . scheme == 'ftp' :
    working_dir = self . dfk . executors [ executor ] . working_dir
    stage_in_app = self . _ftp_stage_in_app ( executor = executor )
    app_fut = stage_in_app ( working_dir , outputs = [ file ] )
    return app_fut . _outputs [ 0 ]
elif file . scheme == 'http' or file . scheme == 'https' :
    working_dir = self . dfk . executors [ executor ] . working_dir
    stage_in_app = self . _http_stage_in_app ( executor = executor )
    app_fut = stage_in_app ( working_dir , outputs = [ file ] )
    return app_fut . _outputs [ 0 ]
elif file . scheme == 'globus' :
    globus_ep = self . _get_globus_endpoint ( executor )
    stage_in_app = self . _globus_stage_in_app ( )
    app_fut = stage_in_app ( globus_ep , outputs = [ file ] )
    return app_fut . _outputs [ 0 ]
else :
    raise Exception ( 'Staging in with unknown file scheme {} is not supported' . format ( file . scheme ) )


signature:
def stage_out ( self , file , executor ) :
    """Transport the file from the local filesystem to the remote Globus endpoint .
 This function returns a DataFuture .
 Args :
 - self
 - file ( File ) - file to stage out
 - executor ( str ) - Which executor the file is going to be staged out from .
 If the executor argument is not specified for a file
 with the ' globus ' scheme , the file will be staged in to
 the first executor with the " globus " key in a config ."""

implementation:
if file . scheme == 'http' or file . scheme == 'https' :
    raise Exception ( 'HTTP/HTTPS file staging out is not supported' )
elif file . scheme == 'ftp' :
    raise Exception ( 'FTP file staging out is not supported' )
elif file . scheme == 'globus' :
    globus_ep = self . _get_globus_endpoint ( executor )
    stage_out_app = self . _globus_stage_out_app ( )
    return stage_out_app ( globus_ep , inputs = [ file ] )
else :
    raise Exception ( 'Staging out with unknown file scheme {} is not supported' . format ( file . scheme ) )


signature:
def get_all_checkpoints ( rundir = "runinfo" ) :
    """Finds the checkpoints from all last runs .
 Note that checkpoints are incremental , and this helper will not find
 previous checkpoints from earlier than the most recent run . It probably
 should be made to do so .
 Kwargs :
 - rundir ( str ) : Path to the runinfo directory
 Returns :
 - a list suitable for the checkpointFiles parameter of DataFlowKernel
 constructor"""

implementation:
if ( not os . path . isdir ( rundir ) ) :
    return [ ]
dirs = sorted ( os . listdir ( rundir ) )
checkpoints = [ ]
for runid in dirs :
    checkpoint = os . path . abspath ( '{}/{}/checkpoint' . format ( rundir , runid ) )
    if os . path . isdir ( checkpoint ) :
        checkpoints . append ( checkpoint )
return checkpoints


signature:
def get_last_checkpoint ( rundir = "runinfo" ) :
    """Find the checkpoint from the last run , if one exists .
 Note that checkpoints are incremental , and this helper will not find
 previous checkpoints from earlier than the most recent run . It probably
 should be made to do so .
 Kwargs :
 - rundir ( str ) : Path to the runinfo directory
 Returns :
 - a list suitable for checkpointFiles parameter of DataFlowKernel
 constructor , with 0 or 1 elements"""

implementation:
if not os . path . isdir ( rundir ) :
    return [ ]
dirs = sorted ( os . listdir ( rundir ) )
if len ( dirs ) == 0 :
    return [ ]
last_runid = dirs [ - 1 ]
last_checkpoint = os . path . abspath ( '{}/{}/checkpoint' . format ( rundir , last_runid ) )
if ( not ( os . path . isdir ( last_checkpoint ) ) ) :
    return [ ]
return [ last_checkpoint ]


signature:
def wtime_to_minutes ( time_string ) :
    '''wtime _ to _ minutes
 Convert standard wallclock time string to minutes .
 Args :
 - Time _ string in HH : MM : SS format
 Returns :
 ( int ) minutes'''

implementation:
hours , mins , seconds = time_string . split ( ':' )
total_mins = int ( hours ) * 60 + int ( mins )
if total_mins < 1 :
    logger . warning ( "Time string '{}' parsed to {} minutes, less than 1" . format ( time_string , total_mins ) )
return total_mins


signature:
def interactive ( f ) :
    """Decorator for making functions appear as interactively defined .
 This results in the function being linked to the user _ ns as globals ( )
 instead of the module globals ( ) ."""

implementation:
# build new FunctionType , so it can have the right globals
# interactive functions never have closures , that ' s kind of the point
if isinstance ( f , FunctionType ) :
    mainmod = __import__ ( '__main__' )
    f = FunctionType ( f . __code__ , mainmod . __dict__ , f . __name__ , f . __defaults__ , )
# associate with _ _ main _ _ for uncanning
f . __module__ = '__main__'
return f


signature:
def use_pickle ( ) :
    """Revert to using stdlib pickle .
 Reverts custom serialization enabled by use _ dill | cloudpickle ."""

implementation:
from . import serialize
serialize . pickle = serialize . _stdlib_pickle
# restore special function handling
can_map [ FunctionType ] = _original_can_map [ FunctionType ]


signature:
def _import_mapping ( mapping , original = None ) :
    """Import any string - keys in a type mapping ."""

implementation:
# log = get _ logger ( )
# log . debug ( " Importing canning map " )
for key , value in list ( mapping . items ( ) ) :
    if isinstance ( key , string_types ) :
        try :
            cls = import_item ( key )
        except Exception :
            if original and key not in original : # only message on user - added classes
            # log . error ( " canning class not importable : % r " , key , exc _ info = True )
                print ( "ERROR: canning class not importable: %r" , key , exc_info = True )
            mapping . pop ( key )
        else :
            mapping [ cls ] = mapping . pop ( key )


signature:
def istype ( obj , check ) :
    """Like isinstance ( obj , check ) , but strict .
 This won ' t catch subclasses ."""

implementation:
if isinstance ( check , tuple ) :
    for cls in check :
        if type ( obj ) is cls :
            return True
    return False
else :
    return type ( obj ) is check


signature:
def can ( obj ) :
    """Prepare an object for pickling ."""

implementation:
import_needed = False
for cls , canner in iteritems ( can_map ) :
    if isinstance ( cls , string_types ) :
        import_needed = True
        break
    elif istype ( obj , cls ) :
        return canner ( obj )
if import_needed : # perform can _ map imports , then try again
# this will usually only happen once
    _import_mapping ( can_map , _original_can_map )
    return can ( obj )
return obj


signature:
def can_dict ( obj ) :
    """Can the * values * of a dict ."""

implementation:
if istype ( obj , dict ) :
    newobj = { }
    for k , v in iteritems ( obj ) :
        newobj [ k ] = can ( v )
    return newobj
else :
    return obj


signature:
def can_sequence ( obj ) :
    """Can the elements of a sequence ."""

implementation:
if istype ( obj , sequence_types ) :
    t = type ( obj )
    return t ( [ can ( i ) for i in obj ] )
else :
    return obj


signature:
def uncan ( obj , g = None ) :
    """Invert canning ."""

implementation:
import_needed = False
for cls , uncanner in iteritems ( uncan_map ) :
    if isinstance ( cls , string_types ) :
        import_needed = True
        break
    elif isinstance ( obj , cls ) :
        return uncanner ( obj , g )
if import_needed : # perform uncan _ map imports , then try again
# this will usually only happen once
    _import_mapping ( uncan_map , _original_uncan_map )
    return uncan ( obj , g )
return obj


signature:
def unset_logging ( self ) :
    """Mute newly added handlers to the root level , right after calling executor . status"""

implementation:
if self . logger_flag is True :
    return
root_logger = logging . getLogger ( )
for hndlr in root_logger . handlers :
    if hndlr not in self . prior_loghandlers :
        hndlr . setLevel ( logging . ERROR )
self . logger_flag = True


signature:
def _strategy_simple ( self , tasks , * args , kind = None , ** kwargs ) :
    """Peek at the DFK and the executors specified .
 We assume here that tasks are not held in a runnable
 state , and that all tasks from an app would be sent to
 a single specific executor , i . e tasks cannot be specified
 to go to one of more executors .
 Args :
 - tasks ( task _ ids ) : Not used here .
 KWargs :
 - kind ( Not used )"""

implementation:
for label , executor in self . dfk . executors . items ( ) :
    if not executor . scaling_enabled :
        continue
    # Tasks that are either pending completion
    active_tasks = executor . outstanding
    status = executor . status ( )
    self . unset_logging ( )
    # FIXME we need to handle case where provider does not define these
    # FIXME probably more of this logic should be moved to the provider
    min_blocks = executor . provider . min_blocks
    max_blocks = executor . provider . max_blocks
    if isinstance ( executor , IPyParallelExecutor ) :
        tasks_per_node = executor . workers_per_node
    elif isinstance ( executor , HighThroughputExecutor ) : # This is probably wrong calculation , we need this to come from the executor
    # since we can ' t know slots ahead of time .
        tasks_per_node = 1
    elif isinstance ( executor , ExtremeScaleExecutor ) :
        tasks_per_node = executor . ranks_per_node
    nodes_per_block = executor . provider . nodes_per_block
    parallelism = executor . provider . parallelism
    running = sum ( [ 1 for x in status if x == 'RUNNING' ] )
    submitting = sum ( [ 1 for x in status if x == 'SUBMITTING' ] )
    pending = sum ( [ 1 for x in status if x == 'PENDING' ] )
    active_blocks = running + submitting + pending
    active_slots = active_blocks * tasks_per_node * nodes_per_block
    if hasattr ( executor , 'connected_workers' ) :
        logger . debug ( 'Executor {} has {} active tasks, {}/{}/{} running/submitted/pending blocks, and {} connected workers' . format ( label , active_tasks , running , submitting , pending , executor . connected_workers ) )
    else :
        logger . debug ( 'Executor {} has {} active tasks and {}/{}/{} running/submitted/pending blocks' . format ( label , active_tasks , running , submitting , pending ) )
    # reset kill timer if executor has active tasks
    if active_tasks > 0 and self . executors [ executor . label ] [ 'idle_since' ] :
        self . executors [ executor . label ] [ 'idle_since' ] = None
    # Case 1
    # No tasks .
    if active_tasks == 0 : # Case 1a
    # Fewer blocks that min _ blocks
        if active_blocks <= min_blocks : # Ignore
        # logger . debug ( " Strategy : Case . 1a " )
            pass
        # Case 1b
        # More blocks than min _ blocks . Scale down
        else : # We want to make sure that max _ idletime is reached
        # before killing off resources
            if not self . executors [ executor . label ] [ 'idle_since' ] :
                logger . debug ( "Executor {} has 0 active tasks; starting kill timer (if idle time exceeds {}s, resources will be removed)" . format ( label , self . max_idletime ) )
                self . executors [ executor . label ] [ 'idle_since' ] = time . time ( )
            idle_since = self . executors [ executor . label ] [ 'idle_since' ]
            if ( time . time ( ) - idle_since ) > self . max_idletime : # We have resources idle for the max duration ,
            # we have to scale _ in now .
                logger . debug ( "Idle time has reached {}s for executor {}; removing resources" . format ( self . max_idletime , label ) )
                executor . scale_in ( active_blocks - min_blocks )
            else :
                pass
                # logger . debug ( " Strategy : Case . 1b . Waiting for timer : { 0 } " . format ( idle _ since ) )
    # Case 2
    # More tasks than the available slots .
    elif ( float ( active_slots ) / active_tasks ) < parallelism : # Case 2a
    # We have the max blocks possible
        if active_blocks >= max_blocks : # Ignore since we already have the max nodes
        # logger . debug ( " Strategy : Case . 2a " )
            pass
        # Case 2b
        else : # logger . debug ( " Strategy : Case . 2b " )
            excess = math . ceil ( ( active_tasks * parallelism ) - active_slots )
            excess_blocks = math . ceil ( float ( excess ) / ( tasks_per_node * nodes_per_block ) )
            logger . debug ( "Requesting {} more blocks" . format ( excess_blocks ) )
            executor . scale_out ( excess_blocks )
    elif active_slots == 0 and active_tasks > 0 : # Case 4
    # Check if slots are being lost quickly ?
        logger . debug ( "Requesting single slot" )
        executor . scale_out ( 1 )
    # Case 3
    # tasks ~ slots
    else : # logger . debug ( " Strategy : Case 3 " )
        pass


signature:
def get_parsl_logger ( logger_name = 'parsl_monitor_logger' , is_logging_server = False , monitoring_config = None , ** kwargs ) :
    """Parameters
 logger _ name : str , optional
 Name of the logger to use . Prevents adding repeat handlers or incorrect handlers
 is _ logging _ server : Bool , optional
 Used internally to determine which handler to return when using local db logging
 monitoring _ config : MonitoringConfig , optional
 Pass in a logger class object to use for generating loggers .
 Returns
 logging . logger object
 Raises
 OptionalModuleMissing"""

implementation:
logger = logging . getLogger ( logger_name )
if monitoring_config is None :
    logger . addHandler ( NullHandler ( ) )
    return logger
if monitoring_config . store is None :
    raise ValueError ( 'No MonitoringStore defined' )
if is_logging_server : # add a handler that will take logs being received on the server and log them to the store
    handler = DatabaseHandler ( monitoring_config . store . connection_string )
    # use the specific name generated by the server or the monitor wrapper
    logger = logging . getLogger ( logger_name )
    logger . setLevel ( logging . INFO )
    logger . addHandler ( handler )
else : # add a handler that will pass logs to the logging server
    handler = RemoteHandler ( monitoring_config . store . logging_server_host , monitoring_config . store . logging_server_port )
    # use the specific name generated by the server or the monitor wrapper
    logger = logging . getLogger ( logger_name )
    logger . setLevel ( logging . INFO )
    logger . addHandler ( handler )
return logger


signature:
def start ( self ) :
    """Start the controller ."""

implementation:
if self . mode == "manual" :
    return
if self . ipython_dir != '~/.ipython' :
    self . ipython_dir = os . path . abspath ( os . path . expanduser ( self . ipython_dir ) )
if self . log :
    stdout = open ( os . path . join ( self . ipython_dir , "{0}.controller.out" . format ( self . profile ) ) , 'w' )
    stderr = open ( os . path . join ( self . ipython_dir , "{0}.controller.err" . format ( self . profile ) ) , 'w' )
else :
    stdout = open ( os . devnull , 'w' )
    stderr = open ( os . devnull , 'w' )
try :
    opts = [ 'ipcontroller' , '' if self . ipython_dir == '~/.ipython' else '--ipython-dir={}' . format ( self . ipython_dir ) , self . interfaces if self . interfaces is not None else '--ip=*' , '' if self . profile == 'default' else '--profile={0}' . format ( self . profile ) , '--reuse' if self . reuse else '' , '--location={}' . format ( self . public_ip ) if self . public_ip else '' , '--port={}' . format ( self . port ) if self . port is not None else '' ]
    if self . port_range is not None :
        opts += [ '--HubFactory.hb={0},{1}' . format ( self . hb_ping , self . hb_pong ) , '--HubFactory.control={0},{1}' . format ( self . control_client , self . control_engine ) , '--HubFactory.mux={0},{1}' . format ( self . mux_client , self . mux_engine ) , '--HubFactory.task={0},{1}' . format ( self . task_client , self . task_engine ) ]
    logger . debug ( "Starting ipcontroller with '{}'" . format ( ' ' . join ( [ str ( x ) for x in opts ] ) ) )
    self . proc = subprocess . Popen ( opts , stdout = stdout , stderr = stderr , preexec_fn = os . setsid )
except FileNotFoundError :
    msg = "Could not find ipcontroller. Please make sure that ipyparallel is installed and available in your env"
    logger . error ( msg )
    raise ControllerError ( msg )
except Exception as e :
    msg = "IPPController failed to start: {0}" . format ( e )
    logger . error ( msg )
    raise ControllerError ( msg )


signature:
def engine_file ( self ) :
    """Specify path to the ipcontroller - engine . json file .
 This file is stored in in the ipython _ dir / profile folders .
 Returns :
 - str , File path to engine file"""

implementation: return os . path . join ( self . ipython_dir , 'profile_{0}' . format ( self . profile ) , 'security/ipcontroller-engine.json' )


signature:
def client_file ( self ) :
    """Specify path to the ipcontroller - client . json file .
 This file is stored in in the ipython _ dir / profile folders .
 Returns :
 - str , File path to client file"""

implementation: return os . path . join ( self . ipython_dir , 'profile_{0}' . format ( self . profile ) , 'security/ipcontroller-client.json' )


signature:
def close ( self ) :
    """Terminate the controller process and its child processes .
 Args :
 - None"""

implementation:
if self . reuse :
    logger . debug ( "Ipcontroller not shutting down: reuse enabled" )
    return
if self . mode == "manual" :
    logger . debug ( "Ipcontroller not shutting down: Manual mode" )
    return
try :
    pgid = os . getpgid ( self . proc . pid )
    os . killpg ( pgid , signal . SIGTERM )
    time . sleep ( 0.2 )
    os . killpg ( pgid , signal . SIGKILL )
    try :
        self . proc . wait ( timeout = 1 )
        x = self . proc . returncode
        if x == 0 :
            logger . debug ( "Controller exited with {0}" . format ( x ) )
        else :
            logger . error ( "Controller exited with {0}. May require manual cleanup" . format ( x ) )
    except subprocess . TimeoutExpired :
        logger . warn ( "Ipcontroller process:{0} cleanup failed. May require manual cleanup" . format ( self . proc . pid ) )
except Exception as e :
    logger . warn ( "Failed to kill the ipcontroller process[{0}]: {1}" . format ( self . proc . pid , e ) )


signature:
def make_hash ( self , task ) :
    """Create a hash of the task inputs .
 This uses a serialization library borrowed from ipyparallel .
 If this fails here , then all ipp calls are also likely to fail due to failure
 at serialization .
 Args :
 - task ( dict ) : Task dictionary from dfk . tasks
 Returns :
 - hash ( str ) : A unique hash string"""

implementation:
# Function name TODO : Add fn body later
t = [ serialize_object ( task [ 'func_name' ] ) [ 0 ] , serialize_object ( task [ 'fn_hash' ] ) [ 0 ] , serialize_object ( task [ 'args' ] ) [ 0 ] , serialize_object ( task [ 'kwargs' ] ) [ 0 ] , serialize_object ( task [ 'env' ] ) [ 0 ] ]
x = b'' . join ( t )
hashedsum = hashlib . md5 ( x ) . hexdigest ( )
return hashedsum


signature:
def check_memo ( self , task_id , task ) :
    """Create a hash of the task and its inputs and check the lookup table for this hash .
 If present , the results are returned . The result is a tuple indicating whether a memo
 exists and the result , since a Null result is possible and could be confusing .
 This seems like a reasonable option without relying on an cache _ miss exception .
 Args :
 - task ( task ) : task from the dfk . tasks table
 Returns :
 Tuple of the following :
 - present ( Bool ) : Is this present in the memo _ lookup _ table
 - Result ( Py Obj ) : Result of the function if present in table
 This call will also set task [ ' hashsum ' ] to the unique hashsum for the func + inputs ."""

implementation:
if not self . memoize or not task [ 'memoize' ] :
    task [ 'hashsum' ] = None
    return None , None
hashsum = self . make_hash ( task )
present = False
result = None
if hashsum in self . memo_lookup_table :
    present = True
    result = self . memo_lookup_table [ hashsum ]
    logger . info ( "Task %s using result from cache" , task_id )
task [ 'hashsum' ] = hashsum
return present , result


signature:
def update_memo ( self , task_id , task , r ) :
    """Updates the memoization lookup table with the result from a task .
 Args :
 - task _ id ( int ) : Integer task id
 - task ( dict ) : A task dict from dfk . tasks
 - r ( Result future ) : Result future
 A warning is issued when a hash collision occurs during the update .
 This is not likely ."""

implementation:
if not self . memoize or not task [ 'memoize' ] :
    return
if task [ 'hashsum' ] in self . memo_lookup_table :
    logger . info ( 'Updating appCache entry with latest %s:%s call' % ( task [ 'func_name' ] , task_id ) )
    self . memo_lookup_table [ task [ 'hashsum' ] ] = r
else :
    self . memo_lookup_table [ task [ 'hashsum' ] ] = r


signature:
def _nbytes ( buf ) :
    """Return byte - size of a memoryview or buffer ."""

implementation:
if isinstance ( buf , memoryview ) :
    if PY3 : # py3 introduces nbytes attribute
        return buf . nbytes
    else : # compute nbytes on py2
        size = buf . itemsize
        for dim in buf . shape :
            size *= dim
        return size
else : # not a memoryview , raw bytes / py2 buffer
    return len ( buf )


signature:
def _extract_buffers ( obj , threshold = MAX_BYTES ) :
    """Extract buffers larger than a certain threshold ."""

implementation:
buffers = [ ]
if isinstance ( obj , CannedObject ) and obj . buffers :
    for i , buf in enumerate ( obj . buffers ) :
        nbytes = _nbytes ( buf )
        if nbytes > threshold : # buffer larger than threshold , prevent pickling
            obj . buffers [ i ] = None
            buffers . append ( buf )
        # buffer too small for separate send , coerce to bytes
        # because pickling buffer objects just results in broken pointers
        elif isinstance ( buf , memoryview ) :
            obj . buffers [ i ] = buf . tobytes ( )
        elif isinstance ( buf , buffer ) :
            obj . buffers [ i ] = bytes ( buf )
return buffers


signature:
def _restore_buffers ( obj , buffers ) :
    """Restore extracted buffers ."""

implementation:
if isinstance ( obj , CannedObject ) and obj . buffers :
    for i , buf in enumerate ( obj . buffers ) :
        if buf is None :
            obj . buffers [ i ] = buffers . pop ( 0 )


signature:
def serialize_object ( obj , buffer_threshold = MAX_BYTES , item_threshold = MAX_ITEMS ) :
    """Serialize an object into a list of sendable buffers .
 Parameters
 obj : object
 The object to be serialized
 buffer _ threshold : int
 The threshold ( in bytes ) for pulling out data buffers
 to avoid pickling them .
 item _ threshold : int
 The maximum number of items over which canning will iterate .
 Containers ( lists , dicts ) larger than this will be pickled without
 introspection .
 Returns
 [ bufs ] : list of buffers representing the serialized object ."""

implementation:
buffers = [ ]
if istype ( obj , sequence_types ) and len ( obj ) < item_threshold :
    cobj = can_sequence ( obj )
    for c in cobj :
        buffers . extend ( _extract_buffers ( c , buffer_threshold ) )
elif istype ( obj , dict ) and len ( obj ) < item_threshold :
    cobj = { }
    for k in sorted ( obj ) :
        c = can ( obj [ k ] )
        buffers . extend ( _extract_buffers ( c , buffer_threshold ) )
        cobj [ k ] = c
else :
    cobj = can ( obj )
    buffers . extend ( _extract_buffers ( cobj , buffer_threshold ) )
buffers . insert ( 0 , pickle . dumps ( cobj , PICKLE_PROTOCOL ) )
return buffers


signature:
def deserialize_object ( buffers , g = None ) :
    """Reconstruct an object serialized by serialize _ object from data buffers .
 Parameters
 bufs : list of buffers / bytes
 g : globals to be used when uncanning
 Returns
 ( newobj , bufs ) : unpacked object , and the list of remaining unused buffers ."""

implementation:
bufs = list ( buffers )
pobj = buffer_to_bytes_py2 ( bufs . pop ( 0 ) )
canned = pickle . loads ( pobj )
if istype ( canned , sequence_types ) and len ( canned ) < MAX_ITEMS :
    for c in canned :
        _restore_buffers ( c , bufs )
    newobj = uncan_sequence ( canned , g )
elif istype ( canned , dict ) and len ( canned ) < MAX_ITEMS :
    newobj = { }
    for k in sorted ( canned ) :
        c = canned [ k ]
        _restore_buffers ( c , bufs )
        newobj [ k ] = uncan ( c , g )
else :
    _restore_buffers ( canned , bufs )
    newobj = uncan ( canned , g )
return newobj , bufs


signature:
def pack_apply_message ( f , args , kwargs , buffer_threshold = MAX_BYTES , item_threshold = MAX_ITEMS ) :
    """Pack up a function , args , and kwargs to be sent over the wire .
 Each element of args / kwargs will be canned for special treatment ,
 but inspection will not go any deeper than that .
 Any object whose data is larger than ` threshold ` will not have their data copied
 ( only numpy arrays and bytes / buffers support zero - copy )
 Message will be a list of bytes / buffers of the format :
 [ cf , pinfo , < arg _ bufs > , < kwarg _ bufs > ]
 With length at least two + len ( args ) + len ( kwargs )"""

implementation:
arg_bufs = list ( chain . from_iterable ( serialize_object ( arg , buffer_threshold , item_threshold ) for arg in args ) )
kw_keys = sorted ( kwargs . keys ( ) )
kwarg_bufs = list ( chain . from_iterable ( serialize_object ( kwargs [ key ] , buffer_threshold , item_threshold ) for key in kw_keys ) )
info = dict ( nargs = len ( args ) , narg_bufs = len ( arg_bufs ) , kw_keys = kw_keys )
msg = [ pickle . dumps ( can ( f ) , PICKLE_PROTOCOL ) ]
msg . append ( pickle . dumps ( info , PICKLE_PROTOCOL ) )
msg . extend ( arg_bufs )
msg . extend ( kwarg_bufs )
return msg


signature:
def unpack_apply_message ( bufs , g = None , copy = True ) :
    """Unpack f , args , kwargs from buffers packed by pack _ apply _ message ( ) .
 Returns : original f , args , kwargs"""

implementation:
bufs = list ( bufs )
# allow us to pop
assert len ( bufs ) >= 2 , "not enough buffers!"
pf = buffer_to_bytes_py2 ( bufs . pop ( 0 ) )
f = uncan ( pickle . loads ( pf ) , g )
pinfo = buffer_to_bytes_py2 ( bufs . pop ( 0 ) )
info = pickle . loads ( pinfo )
arg_bufs , kwarg_bufs = bufs [ : info [ 'narg_bufs' ] ] , bufs [ info [ 'narg_bufs' ] : ]
args = [ ]
for i in range ( info [ 'nargs' ] ) :
    arg , arg_bufs = deserialize_object ( arg_bufs , g )
    args . append ( arg )
args = tuple ( args )
assert not arg_bufs , "Shouldn't be any arg bufs left over"
kwargs = { }
for key in info [ 'kw_keys' ] :
    kwarg , kwarg_bufs = deserialize_object ( kwarg_bufs , g )
    kwargs [ key ] = kwarg
assert not kwarg_bufs , "Shouldn't be any kwarg bufs left over"
return f , args , kwargs


signature:
def _write_submit_script ( self , template , script_filename , job_name , configs ) :
    """Generate submit script and write it to a file .
 Args :
 - template ( string ) : The template string to be used for the writing submit script
 - script _ filename ( string ) : Name of the submit script
 - job _ name ( string ) : job name
 - configs ( dict ) : configs that get pushed into the template
 Returns :
 - True : on success
 Raises :
 SchedulerMissingArgs : If template is missing args
 ScriptPathError : Unable to write submit script out"""

implementation:
try :
    submit_script = Template ( template ) . substitute ( jobname = job_name , ** configs )
    # submit _ script = Template ( template ) . safe _ substitute ( jobname = job _ name , * * configs )
    with open ( script_filename , 'w' ) as f :
        f . write ( submit_script )
except KeyError as e :
    logger . error ( "Missing keys for submit script : %s" , e )
    raise ( SchedulerMissingArgs ( e . args , self . sitename ) )
except IOError as e :
    logger . error ( "Failed writing to submit script: %s" , script_filename )
    raise ( ScriptPathError ( script_filename , e ) )
except Exception as e :
    print ( "Template : " , template )
    print ( "Args : " , job_name )
    print ( "Kwargs : " , configs )
    logger . error ( "Uncategorized error: %s" , e )
    raise ( e )
return True


signature:
def status ( self , job_ids ) :
    """Get the status of a list of jobs identified by the job identifiers
 returned from the submit request .
 Args :
 - job _ ids ( list ) : A list of job identifiers
 Returns :
 - A list of status from [ ' PENDING ' , ' RUNNING ' , ' CANCELLED ' , ' COMPLETED ' ,
 ' FAILED ' , ' TIMEOUT ' ] corresponding to each job _ id in the job _ ids list .
 Raises :
 - ExecutionProviderException or its subclasses"""

implementation:
if job_ids :
    self . _status ( )
return [ self . resources [ jid ] [ 'status' ] for jid in job_ids ]


signature:
def status ( self , job_ids ) :
    '''Get the status of a list of jobs identified by their ids .
 Args :
 - job _ ids ( List of ids ) : List of identifiers for the jobs
 Returns :
 - List of status codes .'''

implementation:
logger . debug ( "Checking status of: {0}" . format ( job_ids ) )
for job_id in self . resources :
    if self . resources [ job_id ] [ 'proc' ] :
        poll_code = self . resources [ job_id ] [ 'proc' ] . poll ( )
        if self . resources [ job_id ] [ 'status' ] in [ 'COMPLETED' , 'FAILED' ] :
            continue
        if poll_code is None :
            self . resources [ job_id ] [ 'status' ] = 'RUNNING'
        elif poll_code == 0 :
            self . resources [ job_id ] [ 'status' ] = 'COMPLETED'
        elif poll_code != 0 :
            self . resources [ job_id ] [ 'status' ] = 'FAILED'
        else :
            logger . error ( "Internal consistency error: unexpected case in local provider state machine" )
    elif self . resources [ job_id ] [ 'remote_pid' ] :
        retcode , stdout , stderr = self . channel . execute_wait ( 'ps -p {} &> /dev/null; echo "STATUS:$?" ' , self . cmd_timeout )
        for line in stdout . split ( '\n' ) :
            if line . startswith ( "STATUS:" ) :
                status = line . split ( "STATUS:" ) [ 1 ] . strip ( )
                if status == "0" :
                    self . resources [ job_id ] [ 'status' ] = 'RUNNING'
                else :
                    self . resources [ job_id ] [ 'status' ] = 'FAILED'
return [ self . resources [ jid ] [ 'status' ] for jid in job_ids ]


signature:
def submit ( self , command , blocksize , tasks_per_node , job_name = "parsl.auto" ) :
    '''Submits the command onto an Local Resource Manager job of blocksize parallel elements .
 Submit returns an ID that corresponds to the task that was just submitted .
 If tasks _ per _ node < 1:
 1 / tasks _ per _ node is provisioned
 If tasks _ per _ node = = 1:
 A single node is provisioned
 If tasks _ per _ node > 1 :
 tasks _ per _ node * blocksize number of nodes are provisioned .
 Args :
 - command : ( String ) Commandline invocation to be made on the remote side .
 - blocksize : ( float ) - Not really used for local
 - tasks _ per _ node ( int ) : command invocations to be launched per node
 Kwargs :
 - job _ name ( String ) : Name for job , must be unique
 Returns :
 - None : At capacity , cannot provision more
 - job _ id : ( string ) Identifier for the job'''

implementation:
job_name = "{0}.{1}" . format ( job_name , time . time ( ) )
# Set script path
script_path = "{0}/{1}.sh" . format ( self . script_dir , job_name )
script_path = os . path . abspath ( script_path )
wrap_command = self . worker_init + '\n' + self . launcher ( command , tasks_per_node , self . nodes_per_block )
self . _write_submit_script ( wrap_command , script_path )
job_id = None
proc = None
remote_pid = None
if ( self . move_files is None and not isinstance ( self . channel , LocalChannel ) ) or ( self . move_files ) :
    logger . debug ( "Moving start script" )
    script_path = self . channel . push_file ( script_path , self . channel . script_dir )
if not isinstance ( self . channel , LocalChannel ) :
    logger . debug ( "Launching in remote mode" )
    # Bash would return until the streams are closed . So we redirect to a outs file
    cmd = 'bash {0} &> {0}.out & \n echo "PID:$!" ' . format ( script_path )
    retcode , stdout , stderr = self . channel . execute_wait ( cmd , self . cmd_timeout )
    for line in stdout . split ( '\n' ) :
        if line . startswith ( "PID:" ) :
            remote_pid = line . split ( "PID:" ) [ 1 ] . strip ( )
            job_id = remote_pid
    if job_id is None :
        logger . warning ( "Channel failed to start remote command/retrieve PID" )
else :
    try :
        job_id , proc = self . channel . execute_no_wait ( 'bash {0}' . format ( script_path ) , self . cmd_timeout )
    except Exception as e :
        logger . debug ( "Channel execute failed for: {}, {}" . format ( self . channel , e ) )
        raise
self . resources [ job_id ] = { 'job_id' : job_id , 'status' : 'RUNNING' , 'blocksize' : blocksize , 'remote_pid' : remote_pid , 'proc' : proc }
return job_id


signature:
def cancel ( self , job_ids ) :
    '''Cancels the jobs specified by a list of job ids
 Args :
 job _ ids : [ < job _ id > . . . ]
 Returns :
 [ True / False . . . ] : If the cancel operation fails the entire list will be False .'''

implementation:
for job in job_ids :
    logger . debug ( "Terminating job/proc_id: {0}" . format ( job ) )
    # Here we are assuming that for local , the job _ ids are the process id ' s
    if self . resources [ job ] [ 'proc' ] :
        proc = self . resources [ job ] [ 'proc' ]
        os . killpg ( os . getpgid ( proc . pid ) , signal . SIGTERM )
        self . resources [ job ] [ 'status' ] = 'CANCELLED'
    elif self . resources [ job ] [ 'remote_pid' ] :
        cmd = "kill -- -$(ps -o pgid={} | grep -o '[0-9]*')" . format ( self . resources [ job ] [ 'remote_pid' ] )
        retcode , stdout , stderr = self . channel . execute_wait ( cmd , self . cmd_timeout )
        if retcode != 0 :
            logger . warning ( "Failed to kill PID: {} and child processes on {}" . format ( self . resources [ job ] [ 'remote_pid' ] , self . label ) )
rets = [ True for i in job_ids ]
return rets


signature:
def submit ( self , command , blocksize , tasks_per_node , job_name = "parsl.auto" ) :
    """Submits the command onto an Local Resource Manager job of blocksize parallel elements .
 Submit returns an ID that corresponds to the task that was just submitted .
 If tasks _ per _ node < 1 : ! This is illegal . tasks _ per _ node should be integer
 If tasks _ per _ node = = 1:
 A single node is provisioned
 If tasks _ per _ node > 1 :
 tasks _ per _ node * blocksize number of nodes are provisioned .
 Args :
 - command : ( String ) Commandline invocation to be made on the remote side .
 - blocksize : ( float )
 - tasks _ per _ node ( int ) : command invocations to be launched per node
 Kwargs :
 - job _ name ( String ) : Name for job , must be unique
 Returns :
 - None : At capacity , cannot provision more
 - job _ id : ( string ) Identifier for the job"""

implementation:
if self . provisioned_blocks >= self . max_blocks :
    logger . warn ( "[%s] at capacity, cannot add more blocks now" , self . label )
    return None
# Note : Fix this later to avoid confusing behavior .
# We should always allocate blocks in integer counts of node _ granularity
if blocksize < self . nodes_per_block :
    blocksize = self . nodes_per_block
account_opt = '-A {}' . format ( self . account ) if self . account is not None else ''
job_name = "parsl.{0}.{1}" . format ( job_name , time . time ( ) )
script_path = "{0}/{1}.submit" . format ( self . script_dir , job_name )
script_path = os . path . abspath ( script_path )
job_config = { }
job_config [ "scheduler_options" ] = self . scheduler_options
job_config [ "worker_init" ] = self . worker_init
logger . debug ( "Requesting blocksize:%s nodes_per_block:%s tasks_per_node:%s" , blocksize , self . nodes_per_block , tasks_per_node )
# Wrap the command
job_config [ "user_script" ] = self . launcher ( command , tasks_per_node , self . nodes_per_block )
queue_opt = '-q {}' . format ( self . queue ) if self . queue is not None else ''
logger . debug ( "Writing submit script" )
self . _write_submit_script ( template_string , script_path , job_name , job_config )
channel_script_path = self . channel . push_file ( script_path , self . channel . script_dir )
command = 'qsub -n {0} {1} -t {2} {3} {4}' . format ( self . nodes_per_block , queue_opt , wtime_to_minutes ( self . walltime ) , account_opt , channel_script_path )
logger . debug ( "Executing {}" . format ( command ) )
retcode , stdout , stderr = super ( ) . execute_wait ( command )
# TODO : FIX this block
if retcode != 0 :
    logger . error ( "Failed command: {0}" . format ( command ) )
    logger . error ( "Launch failed stdout:\n{0} \nstderr:{1}\n" . format ( stdout , stderr ) )
logger . debug ( "Retcode:%s STDOUT:%s STDERR:%s" , retcode , stdout . strip ( ) , stderr . strip ( ) )
job_id = None
if retcode == 0 : # We should be getting only one line back
    job_id = stdout . strip ( )
    self . resources [ job_id ] = { 'job_id' : job_id , 'status' : 'PENDING' , 'blocksize' : blocksize }
else :
    logger . error ( "Submission of command to scale_out failed: {0}" . format ( stderr ) )
    raise ( ScaleOutFailed ( self . __class__ , "Request to submit job to local scheduler failed" ) )
logger . debug ( "Returning job id : {0}" . format ( job_id ) )
return job_id


signature:
def initialize_boto_client ( self ) :
    """Initialize the boto client ."""

implementation:
self . session = self . create_session ( )
self . client = self . session . client ( 'ec2' )
self . ec2 = self . session . resource ( 'ec2' )
self . instances = [ ]
self . instance_states = { }
self . vpc_id = 0
self . sg_id = 0
self . sn_ids = [ ]


signature:
def read_state_file ( self , state_file ) :
    """Read the state file , if it exists .
 If this script has been run previously , resource IDs will have been written to a
 state file . On starting a run , a state file will be looked for before creating new
 infrastructure . Information on VPCs , security groups , and subnets are saved , as
 well as running instances and their states .
 AWS has a maximum number of VPCs per region per account , so we do not want to
 clutter users ' AWS accounts with security groups and VPCs that will be used only
 once ."""

implementation:
try :
    fh = open ( state_file , 'r' )
    state = json . load ( fh )
    self . vpc_id = state [ 'vpcID' ]
    self . sg_id = state [ 'sgID' ]
    self . sn_ids = state [ 'snIDs' ]
    self . instances = state [ 'instances' ]
except Exception as e :
    logger . debug ( "Caught exception while reading state file: {0}" . format ( e ) )
    raise e
logger . debug ( "Done reading state from the local state file." )


signature:
def write_state_file ( self ) :
    """Save information that must persist to a file .
 We do not want to create a new VPC and new identical security groups , so we save
 information about them in a file between runs ."""

implementation:
fh = open ( 'awsproviderstate.json' , 'w' )
state = { }
state [ 'vpcID' ] = self . vpc_id
state [ 'sgID' ] = self . sg_id
state [ 'snIDs' ] = self . sn_ids
state [ 'instances' ] = self . instances
state [ "instanceState" ] = self . instance_states
fh . write ( json . dumps ( state , indent = 4 ) )


signature:
def create_session ( self ) :
    """Create a session .
 First we look in self . key _ file for a path to a json file with the
 credentials . The key file should have ' AWSAccessKeyId ' and ' AWSSecretKey ' .
 Next we look at self . profile for a profile name and try
 to use the Session call to automatically pick up the keys for the profile from
 the user default keys file ~ / . aws / config .
 Finally , boto3 will look for the keys in environment variables :
 AWS _ ACCESS _ KEY _ ID : The access key for your AWS account .
 AWS _ SECRET _ ACCESS _ KEY : The secret key for your AWS account .
 AWS _ SESSION _ TOKEN : The session key for your AWS account .
 This is only needed when you are using temporary credentials .
 The AWS _ SECURITY _ TOKEN environment variable can also be used ,
 but is only supported for backwards compatibility purposes .
 AWS _ SESSION _ TOKEN is supported by multiple AWS SDKs besides python ."""

implementation:
session = None
if self . key_file is not None :
    credfile = os . path . expandvars ( os . path . expanduser ( self . key_file ) )
    try :
        with open ( credfile , 'r' ) as f :
            creds = json . load ( f )
    except json . JSONDecodeError as e :
        logger . error ( "EC2Provider '{}': json decode error in credential file {}" . format ( self . label , credfile ) )
        raise e
    except Exception as e :
        logger . debug ( "EC2Provider '{0}' caught exception while reading credential file: {1}" . format ( self . label , credfile ) )
        raise e
    logger . debug ( "EC2Provider '{}': Using credential file to create session" . format ( self . label ) )
    session = boto3 . session . Session ( region_name = self . region , ** creds )
elif self . profile is not None :
    logger . debug ( "EC2Provider '{}': Using profile name to create session" . format ( self . label ) )
    session = boto3 . session . Session ( profile_name = self . profile , region_name = self . region )
else :
    logger . debug ( "EC2Provider '{}': Using environment variables to create session" . format ( self . label ) )
    session = boto3 . session . Session ( region_name = self . region )
return session


signature:
def create_vpc ( self ) :
    """Create and configure VPC
 We create a VPC with CIDR 10.0.0.0/16 , which provides up to 64,000 instances .
 We attach a subnet for each availability zone within the region specified in the
 config . We give each subnet an ip range like 10.0 . X . 0/20 , which is large enough
 for approx . 4000 instances .
 Security groups are configured in function security _ group ."""

implementation:
try : # We use a large VPC so that the cluster can get large
    vpc = self . ec2 . create_vpc ( CidrBlock = '10.0.0.0/16' , AmazonProvidedIpv6CidrBlock = False , )
except Exception as e : # This failure will cause a full abort
    logger . error ( "{}\n" . format ( e ) )
    raise e
# Attach internet gateway so that our cluster can
# talk to the outside internet
internet_gateway = self . ec2 . create_internet_gateway ( )
internet_gateway . attach_to_vpc ( VpcId = vpc . vpc_id )
# Returns None
self . internet_gateway = internet_gateway . id
# Create and configure route table to allow proper traffic
route_table = self . config_route_table ( vpc , internet_gateway )
self . route_table = route_table . id
# Get all avaliability zones
availability_zones = self . client . describe_availability_zones ( )
# go through AZs and set up a subnet per
for num , zone in enumerate ( availability_zones [ 'AvailabilityZones' ] ) :
    if zone [ 'State' ] == "available" : # Create a large subnet ( 4000 max nodes )
        subnet = vpc . create_subnet ( CidrBlock = '10.0.{}.0/20' . format ( 16 * num ) , AvailabilityZone = zone [ 'ZoneName' ] )
        # Make subnet accessible
        subnet . meta . client . modify_subnet_attribute ( SubnetId = subnet . id , MapPublicIpOnLaunch = { "Value" : True } )
        route_table . associate_with_subnet ( SubnetId = subnet . id )
        self . sn_ids . append ( subnet . id )
    else :
        logger . info ( "{} unavailable" . format ( zone [ 'ZoneName' ] ) )
# Security groups
self . security_group ( vpc )
self . vpc_id = vpc . id
return vpc


signature:
def security_group ( self , vpc ) :
    """Create and configure a new security group .
 Allows all ICMP in , all TCP and UDP in within VPC .
 This security group is very open . It allows all incoming ping requests on all
 ports . It also allows all outgoing traffic on all ports . This can be limited by
 changing the allowed port ranges .
 Parameters
 vpc : VPC instance
 VPC in which to set up security group ."""

implementation:
sg = vpc . create_security_group ( GroupName = "private-subnet" , Description = "security group for remote executors" )
ip_ranges = [ { 'CidrIp' : '10.0.0.0/16' } ]
# Allows all ICMP in , all TCP and UDP in within VPC
in_permissions = [ { 'IpProtocol' : 'TCP' , 'FromPort' : 0 , 'ToPort' : 65535 , 'IpRanges' : ip_ranges , } , { 'IpProtocol' : 'UDP' , 'FromPort' : 0 , 'ToPort' : 65535 , 'IpRanges' : ip_ranges , } , { 'IpProtocol' : 'ICMP' , 'FromPort' : - 1 , 'ToPort' : - 1 , 'IpRanges' : [ { 'CidrIp' : '0.0.0.0/0' } ] , } , { 'IpProtocol' : 'TCP' , 'FromPort' : 22 , 'ToPort' : 22 , 'IpRanges' : [ { 'CidrIp' : '0.0.0.0/0' } ] , } ]
# Allows all TCP out , all TCP and UDP out within VPC
out_permissions = [ { 'IpProtocol' : 'TCP' , 'FromPort' : 0 , 'ToPort' : 65535 , 'IpRanges' : [ { 'CidrIp' : '0.0.0.0/0' } ] , } , { 'IpProtocol' : 'TCP' , 'FromPort' : 0 , 'ToPort' : 65535 , 'IpRanges' : ip_ranges , } , { 'IpProtocol' : 'UDP' , 'FromPort' : 0 , 'ToPort' : 65535 , 'IpRanges' : ip_ranges , } , ]
sg . authorize_ingress ( IpPermissions = in_permissions )
sg . authorize_egress ( IpPermissions = out_permissions )
self . sg_id = sg . id
return sg


signature:
def config_route_table ( self , vpc , internet_gateway ) :
    """Configure route table for Virtual Private Cloud ( VPC ) .
 Parameters
 vpc : dict
 Representation of the VPC ( created by create _ vpc ( ) ) .
 internet _ gateway : dict
 Representation of the internet gateway ( created by create _ vpc ( ) ) ."""

implementation:
route_table = vpc . create_route_table ( )
route_table . create_route ( DestinationCidrBlock = '0.0.0.0/0' , GatewayId = internet_gateway . internet_gateway_id )
return route_table


signature:
def spin_up_instance ( self , command , job_name ) :
    """Start an instance in the VPC in the first available subnet .
 N instances will be started if nodes _ per _ block > 1.
 Not supported . We only do 1 node per block .
 Parameters
 command : str
 Command string to execute on the node .
 job _ name : str
 Name associated with the instances ."""

implementation:
command = Template ( template_string ) . substitute ( jobname = job_name , user_script = command , linger = str ( self . linger ) . lower ( ) , worker_init = self . worker_init )
instance_type = self . instance_type
subnet = self . sn_ids [ 0 ]
ami_id = self . image_id
total_instances = len ( self . instances )
if float ( self . spot_max_bid ) > 0 :
    spot_options = { 'MarketType' : 'spot' , 'SpotOptions' : { 'MaxPrice' : str ( self . spot_max_bid ) , 'SpotInstanceType' : 'one-time' , 'InstanceInterruptionBehavior' : 'terminate' } }
else :
    spot_options = { }
if total_instances > self . max_nodes :
    logger . warn ( "Exceeded instance limit ({}). Cannot continue\n" . format ( self . max_nodes ) )
    return [ None ]
try :
    tag_spec = [ { "ResourceType" : "instance" , "Tags" : [ { 'Key' : 'Name' , 'Value' : job_name } ] } ]
    instance = self . ec2 . create_instances ( MinCount = 1 , MaxCount = 1 , InstanceType = instance_type , ImageId = ami_id , KeyName = self . key_name , SubnetId = subnet , SecurityGroupIds = [ self . sg_id ] , TagSpecifications = tag_spec , InstanceMarketOptions = spot_options , InstanceInitiatedShutdownBehavior = 'terminate' , IamInstanceProfile = { 'Arn' : self . iam_instance_profile_arn } , UserData = command )
except ClientError as e :
    print ( e )
    logger . error ( e . response )
    return [ None ]
except Exception as e :
    logger . error ( "Request for EC2 resources failed : {0}" . format ( e ) )
    return [ None ]
self . instances . append ( instance [ 0 ] . id )
logger . info ( "Started up 1 instance {} . Instance type:{}" . format ( instance [ 0 ] . id , instance_type ) )
return instance


signature:
def shut_down_instance ( self , instances = None ) :
    """Shut down a list of instances , if provided .
 If no instance is provided , the last instance started up will be shut down ."""

implementation:
if instances and len ( self . instances ) > 0 :
    print ( instances )
    try :
        print ( [ i . id for i in instances ] )
    except Exception as e :
        print ( e )
    term = self . client . terminate_instances ( InstanceIds = instances )
    logger . info ( "Shut down {} instances (ids:{}" . format ( len ( instances ) , str ( instances ) ) )
elif len ( self . instances ) > 0 :
    instance = self . instances . pop ( )
    term = self . client . terminate_instances ( InstanceIds = [ instance ] )
    logger . info ( "Shut down 1 instance (id:{})" . format ( instance ) )
else :
    logger . warn ( "No Instances to shut down.\n" )
    return - 1
self . get_instance_state ( )
return term


signature:
def get_instance_state ( self , instances = None ) :
    """Get states of all instances on EC2 which were started by this file ."""

implementation:
if instances :
    desc = self . client . describe_instances ( InstanceIds = instances )
else :
    desc = self . client . describe_instances ( InstanceIds = self . instances )
# pprint . pprint ( desc [ ' Reservations ' ] , indent = 4)
for i in range ( len ( desc [ 'Reservations' ] ) ) :
    instance = desc [ 'Reservations' ] [ i ] [ 'Instances' ] [ 0 ]
    self . instance_states [ instance [ 'InstanceId' ] ] = instance [ 'State' ] [ 'Name' ]
return self . instance_states


signature:
def status ( self , job_ids ) :
    """Get the status of a list of jobs identified by their ids .
 Parameters
 job _ ids : list of str
 Identifiers for the jobs .
 Returns
 list of int
 The status codes of the requsted jobs ."""

implementation:
all_states = [ ]
status = self . client . describe_instances ( InstanceIds = job_ids )
for r in status [ 'Reservations' ] :
    for i in r [ 'Instances' ] :
        instance_id = i [ 'InstanceId' ]
        instance_state = translate_table . get ( i [ 'State' ] [ 'Name' ] , 'UNKNOWN' )
        self . resources [ instance_id ] [ 'status' ] = instance_state
        all_states . extend ( [ instance_state ] )
return all_states


signature:
def submit ( self , command = 'sleep 1' , blocksize = 1 , tasks_per_node = 1 , job_name = "parsl.auto" ) :
    """Submit the command onto a freshly instantiated AWS EC2 instance .
 Submit returns an ID that corresponds to the task that was just submitted .
 Parameters
 command : str
 Command to be invoked on the remote side .
 blocksize : int
 Number of blocks requested .
 tasks _ per _ node : int ( default = 1)
 Number of command invocations to be launched per node
 job _ name : str
 Prefix for the job name .
 Returns
 None or str
 If at capacity , None will be returned . Otherwise , the job identifier will be returned ."""

implementation:
job_name = "parsl.auto.{0}" . format ( time . time ( ) )
wrapped_cmd = self . launcher ( command , tasks_per_node , self . nodes_per_block )
[ instance , * rest ] = self . spin_up_instance ( command = wrapped_cmd , job_name = job_name )
if not instance :
    logger . error ( "Failed to submit request to EC2" )
    return None
logger . debug ( "Started instance_id: {0}" . format ( instance . instance_id ) )
state = translate_table . get ( instance . state [ 'Name' ] , "PENDING" )
self . resources [ instance . instance_id ] = { "job_id" : instance . instance_id , "instance" : instance , "status" : state }
return instance . instance_id


signature:
def cancel ( self , job_ids ) :
    """Cancel the jobs specified by a list of job ids .
 Parameters
 job _ ids : list of str
 List of of job identifiers
 Returns
 list of bool
 Each entry in the list will contain False if the operation fails . Otherwise , the entry will be True ."""

implementation:
if self . linger is True :
    logger . debug ( "Ignoring cancel requests due to linger mode" )
    return [ False for x in job_ids ]
try :
    self . client . terminate_instances ( InstanceIds = list ( job_ids ) )
except Exception as e :
    logger . error ( "Caught error while attempting to remove instances: {0}" . format ( job_ids ) )
    raise e
else :
    logger . debug ( "Removed the instances: {0}" . format ( job_ids ) )
for job_id in job_ids :
    self . resources [ job_id ] [ "status" ] = "COMPLETED"
for job_id in job_ids :
    self . instances . remove ( job_id )
return [ True for x in job_ids ]


signature:
def show_summary ( self ) :
    """Print human readable summary of current AWS state to log and to console ."""

implementation:
self . get_instance_state ( )
status_string = "EC2 Summary:\n\tVPC IDs: {}\n\tSubnet IDs: \
{}\n\tSecurity Group ID: {}\n\tRunning Instance IDs: {}\n" . format ( self . vpc_id , self . sn_ids , self . sg_id , self . instances )
status_string += "\tInstance States:\n\t\t"
self . get_instance_state ( )
for state in self . instance_states . keys ( ) :
    status_string += "Instance ID: {}  State: {}\n\t\t" . format ( state , self . instance_states [ state ] )
status_string += "\n"
logger . info ( status_string )
return status_string


signature:
def teardown ( self ) :
    """Teardown the EC2 infastructure .
 Terminate all EC2 instances , delete all subnets , delete security group , delete VPC ,
 and reset all instance variables ."""

implementation:
self . shut_down_instance ( self . instances )
self . instances = [ ]
try :
    self . client . delete_internet_gateway ( InternetGatewayId = self . internet_gateway )
    self . internet_gateway = None
    self . client . delete_route_table ( RouteTableId = self . route_table )
    self . route_table = None
    for subnet in list ( self . sn_ids ) : # Cast to list ensures that this is a copy
    # Which is important because it means that
    # the length of the list won ' t change during iteration
        self . client . delete_subnet ( SubnetId = subnet )
        self . sn_ids . remove ( subnet )
    self . client . delete_security_group ( GroupId = self . sg_id )
    self . sg_id = None
    self . client . delete_vpc ( VpcId = self . vpc_id )
    self . vpc_id = None
except Exception as e :
    logger . error ( "{}" . format ( e ) )
    raise e
self . show_summary ( )
os . remove ( self . config [ 'state_file_path' ] )


signature:
def scale_out ( self , blocks = 1 , block_size = 1 ) :
    '''Scale out the existing resources .'''

implementation:
self . config [ 'sites.jetstream.{0}' . format ( self . pool ) ] [ 'flavor' ]
count = 0
if blocks == 1 :
    block_id = len ( self . blocks )
    self . blocks [ block_id ] = [ ]
    for instance_id in range ( 0 , block_size ) :
        instances = self . server_manager . create ( 'parsl-{0}-{1}' . format ( block_id , instance_id ) , # Name
        self . client . images . get ( '87e08a17-eae2-4ce4-9051-c561d9a54bde' ) , # Image _ id
        self . client . flavors . list ( ) [ 0 ] , min_count = 1 , max_count = 1 , userdata = setup_script . format ( engine_config = self . engine_config ) , key_name = 'TG-MCB090174-api-key' , security_groups = [ 'global-ssh' ] , nics = [ { "net-id" : '724a50cf-7f11-4b3b-a884-cd7e6850e39e' , "net-name" : 'PARSL-priv-net' , "v4-fixed-ip" : '' } ] )
        self . blocks [ block_id ] . extend ( [ instances ] )
        count += 1
return count


signature:
def scale_in ( self , blocks = 0 , machines = 0 , strategy = None ) :
    '''Scale in resources'''

implementation:
count = 0
instances = self . client . servers . list ( )
for instance in instances [ 0 : machines ] :
    print ( "Deleting : " , instance )
    instance . delete ( )
    count += 1
return count


signature:
def execute_task ( f , args , kwargs , user_ns ) :
    """Deserialize the buffer and execute the task .
 # Returns the result or exception ."""

implementation:
fname = getattr ( f , '__name__' , 'f' )
prefix = "parsl_"
fname = prefix + "f"
argname = prefix + "args"
kwargname = prefix + "kwargs"
resultname = prefix + "result"
user_ns . update ( { fname : f , argname : args , kwargname : kwargs , resultname : resultname } )
code = "{0} = {1}(*{2}, **{3})" . format ( resultname , fname , argname , kwargname )
try :
    exec ( code , user_ns , user_ns )
except Exception as e :
    logger . warning ( "Caught exception; will raise it: {}" . format ( e ) )
    raise e
else :
    return user_ns . get ( resultname )


signature:
def start_file_logger ( filename , rank , name = 'parsl' , level = logging . DEBUG , format_string = None ) :
    """Add a stream log handler .
 Args :
 - filename ( string ) : Name of the file to write logs to
 - name ( string ) : Logger name
 - level ( logging . LEVEL ) : Set the logging level .
 - format _ string ( string ) : Set the format string
 Returns :
 - None"""

implementation:
try :
    os . makedirs ( os . path . dirname ( filename ) , 511 , True )
except Exception as e :
    print ( "Caught exception with trying to make log dirs: {}" . format ( e ) )
if format_string is None :
    format_string = "%(asctime)s %(name)s:%(lineno)d Rank:{0} [%(levelname)s]  %(message)s" . format ( rank )
global logger
logger = logging . getLogger ( name )
logger . setLevel ( logging . DEBUG )
handler = logging . FileHandler ( filename )
handler . setLevel ( level )
formatter = logging . Formatter ( format_string , datefmt = '%Y-%m-%d %H:%M:%S' )
handler . setFormatter ( formatter )
logger . addHandler ( handler )


signature:
def worker ( worker_id , task_url , debug = True , logdir = "workers" , uid = "1" ) :
    """TODO : docstring
 TODO : Cleanup debug , logdir and uid to function correctly"""

implementation:
start_file_logger ( '{}/{}/worker_{}.log' . format ( logdir , uid , worker_id ) , 0 , level = logging . DEBUG if debug is True else logging . INFO )
logger . info ( "Starting worker {}" . format ( worker_id ) )
task_ids_received = [ ]
message_q = zmq_pipes . WorkerMessages ( task_url )
while True :
    print ( "Worker loop iteration starting" )
    task_id , buf = message_q . get ( )
    task_ids_received . append ( task_id )
    user_ns = locals ( )
    user_ns . update ( { '__builtins__' : __builtins__ } )
    f , args , kwargs = unpack_apply_message ( buf , user_ns , copy = False )
    logger . debug ( "Worker {} received task {}" . format ( worker_id , task_id ) )
    result = execute_task ( f , args , kwargs , user_ns )
    logger . debug ( "Worker {} completed task {}" . format ( worker_id , task_id ) )
    reply = { "result" : result , "worker_id" : worker_id }
    message_q . put ( task_id , serialize_object ( reply ) )
    logger . debug ( "Result sent" )