From 9552d34f62db32406aca679853b07edf0f67d14e Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 10 Apr 2020 16:10:16 -0700 Subject: [PATCH 01/27] Check point on threaded local script adapter --- .../datastructures/core/executiongraph.py | 63 +++- maestrowf/datastructures/core/study.py | 4 +- .../script/localparscriptadapter.py | 298 ++++++++++++++++++ maestrowf/maestro.py | 6 +- 4 files changed, 366 insertions(+), 5 deletions(-) create mode 100644 maestrowf/interfaces/script/localparscriptadapter.py diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 632f031f2..a3c9d77ce 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -8,6 +8,7 @@ import os import shutil import tempfile +from concurrent import futures from maestrowf.abstracts.enums import JobStatusCode, State, SubmissionCode, \ CancelCode, StudyStatus @@ -302,7 +303,7 @@ class ExecutionGraph(DAG): """ def __init__(self, submission_attempts=1, submission_throttle=0, - use_tmp=False): + local_procs=1, use_tmp=False): """ Initialize a new instance of an ExecutionGraph. @@ -337,6 +338,8 @@ def __init__(self, submission_attempts=1, submission_throttle=0, self._submission_attempts = submission_attempts self._submission_throttle = submission_throttle + self._local_procs = local_procs + # A map that tracks the dependencies of a step. # NOTE: I don't know how performant the Python dict structure is, but # we'll use it for now. I think this may want to be changed to an AVL @@ -426,6 +429,7 @@ def set_adapter(self, adapter): logger.error(msg) raise TypeError(msg) + print("Adapter set to: {}".format(adapter)) self._adapter = adapter def add_description(self, name, description, **kwargs): @@ -837,6 +841,15 @@ def execute_ready_steps(self): # We now have a collection of ready steps. Execute. # If we don't have a submission limit, go ahead and submit all. + # Check requested resources -> nprocs + nthreads = 1 + if self._local_procs > 0: + nthreads = self._local_procs + + if adapter.total_procs != nthreads: + adapter.total_procs = nthreads + adapter.avail_procs = nthreads + if self._submission_throttle == 0: logger.info("Launching all ready steps...") _available = len(self.ready_steps) @@ -851,8 +864,49 @@ def execute_ready_steps(self): # computed number of slots. We could have free slots, but have less # in the queue. _available = min(_available, len(self.ready_steps)) + logger.info("Found %d available slots...", _available) + + + # print("Num threads: {}".format(nthreads)) + # with futures.ThreadPoolExecutor(max_workers=nthreads) as executor: + # steps_to_do = [] + # for i in range(0, _available): + # # Pop the record and execute using the helper method. + # _record = self.values[self.ready_steps.popleft()] + + # # If we get to this point and we've cancelled, cancel the record. + # if self.is_canceled: + # logger.info("Cancelling '%s' -- continuing.", _record.name) + # _record.mark_end(State.CANCELLED) + # self.cancelled_steps.add(_record.name) + # continue + + # logger.debug("Launching job %d -- %s", i, _record.name) + # steps_to_do.append(executor.submit(self._execute_record, _record, adapter)) + + # for step_future in futures.as_completed(steps_to_do): + # if not step_future: + # print("Encountered None instead of a future.") + # else: + # res = step_future.result() + + + # for i in range(0, _available): + # # Pop the record and execute using the helper method. + # _record = self.values[self.ready_steps.popleft()] + + # # If we get to this point and we've cancelled, cancel the record. + # if self.is_canceled: + # logger.info("Cancelling '%s' -- continuing.", _record.name) + # _record.mark_end(State.CANCELLED) + # self.cancelled_steps.add(_record.name) + # continue + + # logger.debug("Launching job %d -- %s", i, _record.name) + # self._execute_record(_record, adapter) + for i in range(0, _available): # Pop the record and execute using the helper method. _record = self.values[self.ready_steps.popleft()] @@ -864,8 +918,11 @@ def execute_ready_steps(self): self.cancelled_steps.add(_record.name) continue - logger.debug("Launching job %d -- %s", i, _record.name) - self._execute_record(_record, adapter) + + avail_procs = adapter.avail_procs + if _record.step._procs <= avail_procs: + logger.debug("Launching job %d -- %s", i, _record.name) + self._execute_record(_record, adapter) # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/datastructures/core/study.py b/maestrowf/datastructures/core/study.py index c9602ec26..b2535a8c0 100644 --- a/maestrowf/datastructures/core/study.py +++ b/maestrowf/datastructures/core/study.py @@ -390,7 +390,7 @@ def setup_environment(self): self.environment.acquire_environment() def configure_study(self, submission_attempts=1, restart_limit=1, - throttle=0, use_tmp=False, hash_ws=False): + throttle=0, use_tmp=False, hash_ws=False, local_procs=1): """ Perform initial configuration of a study. \ @@ -411,6 +411,7 @@ def configure_study(self, submission_attempts=1, restart_limit=1, self._submission_attempts = submission_attempts self._restart_limit = restart_limit self._submission_throttle = throttle + self._local_procs = local_procs self._use_tmp = use_tmp self._hash_ws = hash_ws @@ -828,6 +829,7 @@ def stage(self): dag = ExecutionGraph( submission_attempts=self._submission_attempts, submission_throttle=self._submission_throttle, + local_procs=self._local_procs, use_tmp=self._use_tmp) dag.add_description(**self.description) dag.log_description() diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py new file mode 100644 index 000000000..6decf85c2 --- /dev/null +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -0,0 +1,298 @@ +############################################################################### +# Copyright (c) 2017, Lawrence Livermore National Security, LLC. +# Produced at the Lawrence Livermore National Laboratory +# Written by Francesco Di Natale, dinatale3@llnl.gov. +# +# LLNL-CODE-734340 +# All rights reserved. +# This file is part of MaestroWF, Version: 1.0.0. +# +# For details, see https://github.com/LLNL/maestrowf. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +############################################################################### + +"""Local interface implementation.""" +import logging +import os +from concurrent.futures import ThreadPoolExecutor +from queue import Queue # priority queue too? +from threading import Thread + +from maestrowf.abstracts.enums import JobStatusCode, SubmissionCode, \ + CancelCode +from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord +from maestrowf.abstracts.interfaces import ScriptAdapter +from maestrowf.abstracts import Singleton +from maestrowf.utils import start_process + +LOGGER = logging.getLogger(__name__) + + +class LocalParallelScriptAdapter(ScriptAdapter): + """A ScriptAdapter class for interfacing for parallel local execution.""" + + key = "local_parallel" + + executor = None + scheduler_thread = None + running_steps = {} # needs to be singleton, threadsafe + submit_queue = None + total_procs = 1 + avail_procs = 1 + + def __init__(self, **kwargs): + """ + Initialize an instance of the LocalParallelScriptAdapter. + + The LocalParallelScriptAdapter is the adapter that is used for workflows that + will execute on the user's machine. This adapter constructs shell scripts for + a StudyStep based on user set defaults and local settings present in each step. + + These scripts are submitted to a local scheduler for asynchronous execution + and monitoring. + + :param **kwargs: A dictionary with default settings for the adapter. + """ + LOGGER.debug("kwargs\n--------------------------\n%s", kwargs) + super(LocalParallelScriptAdapter, self).__init__(**kwargs) + + # Register keys + self.add_batch_parameter("procs", kwargs.pop("procs", "1")) + self.add_batch_parameter("walltime") + + self._header = { + "procs": "# procs = {procs}", + } + + # Maybe need better types here for populating the queues? + self.START = 0 + self.CANCEL = 1 + self.DONE = 2 + + self.total_procs = kwargs.pop("proc_count", "1") + self.avail_procs = self.total_procs + self.submit_queue = Queue() + self.submit_queue.put(self.START) + self.executor = ThreadPoolExecutor(max_workers=self.total_procs) + self.scheduler_thread = Thread(target=self._scheduler_loop()) + + + def _write_script(self, ws_path, step): + """ + Write a shell script to the workspace of a workflow step. + + The job_map optional parameter is a map of workflow step names to job + identifiers. This parameter so far is only planned to be used when a + study is configured to be launched in one go (more or less a script + chain using a scheduler's dependency setting). The functionality of + the parameter may change depending on both future intended use. + + :param ws_path: Path to the workspace directory of the step. + :param step: An instance of a StudyStep. + :returns: False (will not be scheduled), the path to the + written script for run["cmd"], and the path to the script written + for run["restart"] (if it exists). + """ + cmd = step.run["cmd"] + restart = step.run["restart"] + to_be_scheduled = False + + fname = "{}.sh".format(step.name) + script_path = os.path.join(ws_path, fname) + with open(script_path, "w") as script: + script.write("#!{0}\n\n{1}\n".format(self._exec, cmd)) + + if restart: + rname = "{}.restart.sh".format(step.name) + restart_path = os.path.join(ws_path, rname) + + with open(restart_path, "w") as script: + script.write("#!{0}\n\n{1}\n".format(self._exec, restart)) + else: + restart_path = None + + return to_be_scheduled, script_path, restart_path + + def check_jobs(self, joblist): + """ + For the given job list, query execution status. + + This method uses the scontrol show job command and does a + regex search for job information. + + :param joblist: A list of job identifiers to be queried. + :returns: The return code of the status query, and a dictionary of job + identifiers to their status. + """ + + status = {} + status_code = JobStatusCode.OK + + for fut in joblist: # Note, + LOGGER.debug("Looking for job with Future id %s", fut.id) + + # Future states: running, done, cancelled + if fut.done(): + # Check result/process retcode: subprocess errors caught here + if fut in self.running_steps: + result=fut.get_result() # this the right place to catch this? + + status[fut.id] = State.FINISHING + + else: + status[fut.id] = State.FINISHED + + elif fut.running(): + status[fut.id] = State.RUNNING + + elif fut.cancelled(): + # What about cancelled pid's? that would show up under fut.done()... + status[fut.id] = State.CANCELLED + + else: + status[fut.id] = State.UNKNOWN # this ever reached, and if so how + status_code = JobStatusCode.ERROR + + return status_code, status + + def cancel_jobs(self, joblist): + """ + For the given job list, cancel each job. + + :param joblist: A list of job identifiers to be cancelled. + :returns: The return code to indicate if jobs were cancelled. + """ + if not joblist: + return CancellationRecord(CancelCode.OK, 0) + + retcode = 0 + + for fut in joblist: + if fut in self.running_steps: + if fut.done(): # cleanup + # Really need _submit to return submission record vs retcode/err? + result = fut.get_result() + else: + # Interrupt running subprocesses + try: + process, procs = self.running_steps[fut] + process.kill() # better way to kill it? + fut.get_result() # wait on future to exit + except KeyError: + LOGGER.error("Error, future {}, no longer in running step list".format(fut)) + retcode += 1 + + except: # TODO: catch exceptions from kill(), get_result() + LOGGER.error("Error, unexpected behavior trying to cancel future {}, " + "and subprocess with id {}".format(fut, process.pid)) + retcode += 1 + + if not fut.done() and not fut.running(): + # This shouldn't be hit since execution graph only submits if resources available + LOGGER.error("Error, encounterd future {} in unexpected state".format(fut)) + retcode += 1 + + if retcode == 0: + _record = CancellationRecord(CancelCode.OK, retcode) + else: + LOGGER.error("Error code '%s' seen. Unexpected behavior " + "encountered.") # NOTE: what does logger.error inject in %s? + _record = CancellationRecord(CancelCode.ERROR, retcode) + + return _record + + def _submit(self, p, step, path, cwd, job_map=None, env=None): + """ + Execute the step locally. + + If cwd is specified, the submit method will operate outside of the path + specified by the 'cwd' parameter. + If env is specified, the submit method will set the environment + variables for submission to the specified values. The 'env' parameter + should be a dictionary of environment variables. + + :param p: Subprocess object + :param step: An instance of a StudyStep. + :param path: Path to the script to be executed. + :param cwd: Path to the current working directory. + :param job_map: A map of workflow step names to their job identifiers. + :param env: A dict containing a modified environment for execution. + :returns: The return code of the submission command and job identiifer. + """ + LOGGER.debug("cwd = %s", cwd) + LOGGER.debug("Script to execute: %s", path) + + pid = p.pid + output, err = p.communicate() + retcode = p.wait() + + o_path = os.path.join(cwd, "{}.out".format(step.name)) + e_path = os.path.join(cwd, "{}.err".format(step.name)) + + with open(o_path, "w") as out: + out.write(output) + + with open(e_path, "w") as out: + out.write(err) + + if retcode == 0: + LOGGER.info("Execution returned status OK.") + return SubmissionRecord(SubmissionCode.OK, retcode, pid) + else: + LOGGER.warning("Execution returned an error: %s", str(err)) + _record = SubmissionRecord(SubmissionCode.ERROR, retcode, pid) + _record.add_info("stderr", str(err)) + return _record + + def submit(self, step, path, cwd, job_map=None, env=None): + """ + Execute the step locally. + + If cwd is specified, the submit method will operate outside of the path + specified by the 'cwd' parameter. + If env is specified, the submit method will set the environment + variables for submission to the specified values. The 'env' parameter + should be a dictionary of environment variables. + + :param step: An instance of a StudyStep. + :param path: Path to the script to be executed. + :param cwd: Path to the current working directory. + :param job_map: A map of workflow step names to their job identifiers. + :param env: A dict containing a modified environment for execution. + :returns: The return code of the submission command and job identiifer. + """ + try: + p = start_process(path, shell=False, cwd=cwd, env=env) + fut = executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + + except: # add some relevant exceptions here.. + LOGGER.warning("Execution returned an error") + + _record = SubmissionRecord(SubmissionCode.ERROR, -1) + self.avail_procs += step._procs + return _record + + # self.avail_procs -= step._procs + # except: + + # Update running task list. Thread safe list be better? + self.running_steps[fut] = (p, step._procs) + + LOGGER.info("Execution returned status OK.") + return SubmissionRecord(SubmissionCode.OK, 0, fut) diff --git a/maestrowf/maestro.py b/maestrowf/maestro.py index 0e5878d2d..52fdc318a 100644 --- a/maestrowf/maestro.py +++ b/maestrowf/maestro.py @@ -236,7 +236,8 @@ def run_study(args): study.setup_environment() study.configure_study( throttle=args.throttle, submission_attempts=args.attempts, - restart_limit=args.rlimit, use_tmp=args.usetmp, hash_ws=args.hashws) + restart_limit=args.rlimit, use_tmp=args.usetmp, hash_ws=args.hashws, + local_procs=args.local_procs) # Stage the study. path, exec_dag = study.stage() @@ -332,6 +333,9 @@ def setup_argparser(): help="Maximum number of inflight jobs allowed to execute " "simultaneously (0 denotes not throttling)." "[Default: %(default)d]") + run.add_argument("-lp", "--local_procs", type=int, default=1, + help="Max number of processors to use simultaneously." + "[Default: %(default)d]") run.add_argument("-s", "--sleeptime", type=int, default=60, help="Amount of time (in seconds) for the manager to " "wait between job status checks. [Default: %(default)d]") From 0fe1b350c6422fae51e98b09d41459d74f1c94e5 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 10 Apr 2020 19:36:35 -0700 Subject: [PATCH 02/27] Fixes and hacks to get demo spec running. --- .../datastructures/core/executiongraph.py | 14 +- maestrowf/interfaces/__init__.py | 1 + .../script/localparscriptadapter.py | 123 +++++++++++++++--- 3 files changed, 119 insertions(+), 19 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 8f4ac5b1f..7a52ac878 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -426,6 +426,7 @@ def set_adapter(self, adapter): msg = "'{}' adapter must be specfied in ScriptAdapterFactory." \ .format(adapter) LOGGER.error(msg) + LOGGER.error("Valid adapters: {}".format(ScriptAdapterFactory.get_valid_adapters())) raise TypeError(msg) print("Adapter set to: {}".format(adapter)) @@ -882,7 +883,18 @@ def execute_ready_steps(self): continue avail_procs = adapter.avail_procs - if _record.step._procs <= avail_procs: + step_procs = _record.step.run.get("procs") + if not step_procs: + step_procs = 1 + else: + try: + step_procs = int(step_procs) + except ValueError: + step_procs = 1 + LOGGER.error("Setting step {} with no 'procs' attribute to use 1 processor.".format(_record.step.name)) + + # NOTE: better place to set this default, and do type conversions? + if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) diff --git a/maestrowf/interfaces/__init__.py b/maestrowf/interfaces/__init__.py index 156760a6f..671e2b1b8 100644 --- a/maestrowf/interfaces/__init__.py +++ b/maestrowf/interfaces/__init__.py @@ -52,6 +52,7 @@ def iter_adapters(): mods = [(name, ispkg) for finder, name, ispkg in pkgutil.iter_modules( loader.load_module('maestrowf.interfaces.script').__path__, loader.load_module('maestrowf.interfaces.script').__name__ + ".")] + cs = [] for name, _ in mods: # get loader for every module diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 6decf85c2..600e99f37 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -31,28 +31,28 @@ import logging import os from concurrent.futures import ThreadPoolExecutor -from queue import Queue # priority queue too? -from threading import Thread +from functools import partial as func_partial +from threading import Thread, RLock from maestrowf.abstracts.enums import JobStatusCode, SubmissionCode, \ - CancelCode + CancelCode, State from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord -from maestrowf.abstracts.interfaces import ScriptAdapter +from maestrowf.abstracts.interfaces import SchedulerScriptAdapter from maestrowf.abstracts import Singleton from maestrowf.utils import start_process + LOGGER = logging.getLogger(__name__) -class LocalParallelScriptAdapter(ScriptAdapter): +class LocalParallelScriptAdapter(SchedulerScriptAdapter): """A ScriptAdapter class for interfacing for parallel local execution.""" key = "local_parallel" executor = None - scheduler_thread = None running_steps = {} # needs to be singleton, threadsafe - submit_queue = None + tasks_lock = RLock() total_procs = 1 avail_procs = 1 @@ -73,8 +73,9 @@ def __init__(self, **kwargs): super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys - self.add_batch_parameter("procs", kwargs.pop("procs", "1")) - self.add_batch_parameter("walltime") + self.add_batch_parameter("procs", int(kwargs.pop("procs", "1"))) + #self.add_batch_parameter("walltime", kwargs.pop("walltime", ) + # NOTE: add walltime, use this as a timeout on the futures? self._header = { "procs": "# procs = {procs}", @@ -85,14 +86,83 @@ def __init__(self, **kwargs): self.CANCEL = 1 self.DONE = 2 - self.total_procs = kwargs.pop("proc_count", "1") + self.total_procs = int(kwargs.pop("proc_count", "1")) self.avail_procs = self.total_procs - self.submit_queue = Queue() - self.submit_queue.put(self.START) self.executor = ThreadPoolExecutor(max_workers=self.total_procs) - self.scheduler_thread = Thread(target=self._scheduler_loop()) + #self.scheduler_thread = Thread(target=self._scheduler_loop()) + + def _update_running_steps(self, add_tasks=None, rm_tasks=None): + """Thread safe addition/removal of tasks from running_steps dict""" + if not add_tasks: + add_tasks = [] + + if not rm_tasks: + rm_tasks = [] + + with self.tasks_lock: + for task_id in rm_tasks: + if task_id in self.running_steps: + LOGGER.debug("Removing task {} from running_steps".format( + self.running_steps.pop(task_id))) + else: + LOGGER.error("Tried removing task {} from running_steps, " + "but it was not found.".format(task_id)) + + for task in add_tasks: + for task_id in task: + if task_id in self.running_steps: + LOGGER.error("Tried adding already tracked task {} " + "to running_steps.".format(task_id)) + else: + LOGGER.debug("Adding task {} to running_steps".format(task_id)) + self.running_steps[task_id] = task[task_id] + + # Want any status codes returned here if errors triggered? + + def get_header(self, step): + """ + Generate the header present at the top of execution scripts. + + :param step: A StudyStep instance. + :returns: A string of the header based on internal batch parameters and + the parameter step. + """ + return "" + + def get_parallelize_command(self, procs, nodes, **kwargs): + """ + Generate the parallelization segement of the command line. + + :param procs: Number of processors to allocate to the parallel call. + :param nodes: Number of nodes to allocate to the parallel call + (default = 1). + :returns: A string of the parallelize command configured using nodes + and procs. + :NOTE: this is currently a dummy method -> rework to add user specified replacements later + """ + return "" + + def _state(self, fut): + """ + Map a scheduler specific job state to a Study.State enum. + + :param fut: Future instance representing a task + :returns: A Study.State enum corresponding to parameter job_state. + """ + # NOTE: should all of this replace what's in check_jobs -> what about error code there? + if fut.running(): + return State.RUNNING + elif fut.done(): + if fut in self.running_steps: + return State.FINISHING + else: + return State.FINISHED + elif fut.cancelled(): + return State.CANCELLED + + else: + return State.UNKNOWN - def _write_script(self, ws_path, step): """ Write a shell script to the workspace of a workflow step. @@ -109,9 +179,10 @@ def _write_script(self, ws_path, step): written script for run["cmd"], and the path to the script written for run["restart"] (if it exists). """ + # THIS IS A HACK FOR NOW: make better use of get_scheduler_command later cmd = step.run["cmd"] restart = step.run["restart"] - to_be_scheduled = False + to_be_scheduled, _, _ = self.get_scheduler_command(step) fname = "{}.sh".format(step.name) script_path = os.path.join(ws_path, fname) @@ -279,20 +350,36 @@ def submit(self, step, path, cwd, job_map=None, env=None): """ try: p = start_process(path, shell=False, cwd=cwd, env=env) - fut = executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + fut = self.executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + try: + step_procs = step.run.get("procs") + if not step_procs: + step_procs = 1 + else: + step_procs = int(step_procs) + except ValueError: + step_procs = 1 + LOGGER.error("Starting step {} with no 'procs' attribute with 1 processor.".format(_record.step.name)) + + self.avail_procs -= step_procs except: # add some relevant exceptions here.. LOGGER.warning("Execution returned an error") _record = SubmissionRecord(SubmissionCode.ERROR, -1) - self.avail_procs += step._procs + self.avail_procs += step_procs return _record # self.avail_procs -= step._procs # except: # Update running task list. Thread safe list be better? - self.running_steps[fut] = (p, step._procs) + + self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) + fut.add_done_callback(func_partial(self._task_callback, rm_func=self._update_running_steps)) LOGGER.info("Execution returned status OK.") return SubmissionRecord(SubmissionCode.OK, 0, fut) + + def _task_callback(fut, rm_func): + rm_func(add_tasks=None, rm_tasks=[fut]) From 92709b149230ab32c9fa97d1df3dd3e926e07b01 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 13 Apr 2020 18:26:36 -0700 Subject: [PATCH 03/27] Update future callbacks, exclude threadpool from pickling --- .../script/localparscriptadapter.py | 44 +++++++++++++++---- 1 file changed, 35 insertions(+), 9 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 600e99f37..1ba596539 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -91,6 +91,17 @@ def __init__(self, **kwargs): self.executor = ThreadPoolExecutor(max_workers=self.total_procs) #self.scheduler_thread = Thread(target=self._scheduler_loop()) + def __getstate__(self): + """Helper for excluding threadpool from pickling""" + state = self.__dict__.copy() + del state["executor"] + return state + + def __setstate__(self, state): + self.__dict__.update(state) + # Add baz back since it doesn't exist in the pickle + self.executor = ThreadPoolExecutor(max_workers=self.total_procs) + def _update_running_steps(self, add_tasks=None, rm_tasks=None): """Thread safe addition/removal of tasks from running_steps dict""" if not add_tasks: @@ -102,6 +113,8 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): with self.tasks_lock: for task_id in rm_tasks: if task_id in self.running_steps: + self.avail_procs += self.running_steps[task_id][1] # give back resources + LOGGER.debug("Removing task {} from running_steps".format( self.running_steps.pop(task_id))) else: @@ -215,29 +228,31 @@ def check_jobs(self, joblist): status = {} status_code = JobStatusCode.OK - for fut in joblist: # Note, - LOGGER.debug("Looking for job with Future id %s", fut.id) + for fut in joblist: + LOGGER.debug("Looking for job with Future %s", fut) # Future states: running, done, cancelled if fut.done(): # Check result/process retcode: subprocess errors caught here if fut in self.running_steps: - result=fut.get_result() # this the right place to catch this? + # NOTE: does the avail_procs in update_running_steps always catch ending or + # is there some other check needed here? + result=fut.result() # this the right place to catch this? - status[fut.id] = State.FINISHING + status[fut] = State.FINISHING else: - status[fut.id] = State.FINISHED + status[fut] = State.FINISHED elif fut.running(): - status[fut.id] = State.RUNNING + status[fut] = State.RUNNING elif fut.cancelled(): # What about cancelled pid's? that would show up under fut.done()... - status[fut.id] = State.CANCELLED + status[fut] = State.CANCELLED else: - status[fut.id] = State.UNKNOWN # this ever reached, and if so how + status[fut] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR return status_code, status @@ -376,10 +391,21 @@ def submit(self, step, path, cwd, job_map=None, env=None): # Update running task list. Thread safe list be better? self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) - fut.add_done_callback(func_partial(self._task_callback, rm_func=self._update_running_steps)) + # fut.add_done_callback(func_partial(self._task_callback, fut, rm_func=self._update_running_steps)) + fut.add_done_callback(self.wrap_callback(self._update_running_steps)) LOGGER.info("Execution returned status OK.") return SubmissionRecord(SubmissionCode.OK, 0, fut) + @staticmethod + def wrap_callback(rm_func): + def task_callback(fut): + fut.result() + rm_func(add_tasks=None, rm_tasks=[fut]) + + return task_callback + + @staticmethod def _task_callback(fut, rm_func): + fut.get_result() rm_func(add_tasks=None, rm_tasks=[fut]) From 5c04a33a107b20a560351cc540fe1fde6834099b Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Wed, 15 Apr 2020 00:15:21 -0700 Subject: [PATCH 04/27] Some dirty hacks to deal with hidden pending state. --- .../interfaces/script/localparscriptadapter.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 1ba596539..ae3b2b542 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -31,6 +31,8 @@ import logging import os from concurrent.futures import ThreadPoolExecutor +from concurrent.futures._base import PENDING as future_PENDING + from functools import partial as func_partial from threading import Thread, RLock @@ -172,7 +174,8 @@ def _state(self, fut): return State.FINISHED elif fut.cancelled(): return State.CANCELLED - + elif fut._state == future_PENDING: + return State.PENDING else: return State.UNKNOWN @@ -238,7 +241,7 @@ def check_jobs(self, joblist): # NOTE: does the avail_procs in update_running_steps always catch ending or # is there some other check needed here? result=fut.result() # this the right place to catch this? - + LOGGER.debug("Fut {} has result: {}".format(fut, result)) status[fut] = State.FINISHING else: @@ -251,7 +254,16 @@ def check_jobs(self, joblist): # What about cancelled pid's? that would show up under fut.done()... status[fut] = State.CANCELLED + elif fut._state == future_PENDING: + LOGGER.debug("Fut {} is pending".format(fut)) + status[fut] = State.PENDING + else: + print("Fut {} with unknown state".format(fut)) + if fut in self.running_steps: + LOGGER.debug("Fut {} is running?".format(fut)) + LOGGER.debug("Fut {} state: running {}, done {}, cancelled {}".format( + fut, fut.done(), fut.running(), fut.cancelled())) status[fut] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR @@ -400,7 +412,7 @@ def submit(self, step, path, cwd, job_map=None, env=None): @staticmethod def wrap_callback(rm_func): def task_callback(fut): - fut.result() + #fut.result() rm_func(add_tasks=None, rm_tasks=[fut]) return task_callback From 04906cdfd5e37e7cdc255282d707086319b774a6 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 20 Apr 2020 14:36:43 -0700 Subject: [PATCH 05/27] Remove unneeded hooks for command line local_procs parameter --- maestrowf/datastructures/core/executiongraph.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 7a52ac878..2084504c4 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -807,13 +807,13 @@ def execute_ready_steps(self): # We now have a collection of ready steps. Execute. # If we don't have a submission limit, go ahead and submit all. # Check requested resources -> nprocs - nthreads = 1 - if self._local_procs > 0: - nthreads = self._local_procs + # nthreads = 1 + # if self._local_procs > 0: + # nthreads = self._local_procs - if adapter.total_procs != nthreads: - adapter.total_procs = nthreads - adapter.avail_procs = nthreads + # if adapter.total_procs != nthreads: + # adapter.total_procs = nthreads + # adapter.avail_procs = nthreads if self._submission_throttle == 0: LOGGER.info("Launching all ready steps...") @@ -882,6 +882,8 @@ def execute_ready_steps(self): self.cancelled_steps.add(_record.name) continue + # NOTE: verify this actually updates avail_procs on the fly, thus allowing the + # available tasks to be fully consumed before going back to sleep avail_procs = adapter.avail_procs step_procs = _record.step.run.get("procs") if not step_procs: From b9892a9931442a25bcb74327e4d356df6025f762 Mon Sep 17 00:00:00 2001 From: Jeff Greenough Date: Fri, 24 Apr 2020 08:47:34 -0700 Subject: [PATCH 06/27] correctly set avail_procs and total_procs for local parallel adapter --- maestrowf/datastructures/core/executiongraph.py | 4 ++++ maestrowf/interfaces/script/localparscriptadapter.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 2084504c4..3ae17c054 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -885,6 +885,8 @@ def execute_ready_steps(self): # NOTE: verify this actually updates avail_procs on the fly, thus allowing the # available tasks to be fully consumed before going back to sleep avail_procs = adapter.avail_procs + LOGGER.debug("avail_procs from the adapter = %d", avail_procs) + LOGGER.debug("total_procs from the adapter = %d", adapter.total_procs) step_procs = _record.step.run.get("procs") if not step_procs: step_procs = 1 @@ -899,6 +901,8 @@ def execute_ready_steps(self): if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) + else: + LOGGER.debug("step_procs thought > avail_procs: %d : %d", step_procs, avail_procs) # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index ae3b2b542..26a724156 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -89,7 +89,7 @@ def __init__(self, **kwargs): self.DONE = 2 self.total_procs = int(kwargs.pop("proc_count", "1")) - self.avail_procs = self.total_procs + self.avail_procs = int(kwargs.pop("avail_procs", "1")) self.executor = ThreadPoolExecutor(max_workers=self.total_procs) #self.scheduler_thread = Thread(target=self._scheduler_loop()) From 6daa109c8479de3092ff1227b6cfd8c3451d2f1e Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 24 Apr 2020 11:46:11 -0700 Subject: [PATCH 07/27] Enable cancellation of processes, fix cancel status marking --- .../datastructures/core/executiongraph.py | 4 +- .../script/localparscriptadapter.py | 48 +++++++++++++------ 2 files changed, 37 insertions(+), 15 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 2084504c4..d497a74ff 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -429,7 +429,6 @@ def set_adapter(self, adapter): LOGGER.error("Valid adapters: {}".format(ScriptAdapterFactory.get_valid_adapters())) raise TypeError(msg) - print("Adapter set to: {}".format(adapter)) self._adapter = adapter def add_description(self, name, description, **kwargs): @@ -818,6 +817,7 @@ def execute_ready_steps(self): if self._submission_throttle == 0: LOGGER.info("Launching all ready steps...") _available = len(self.ready_steps) + LOGGER.info("Ready steps: {}".format(self.ready_steps)) # Else, we have a limit -- adhere to it. else: # Compute the number of available slots we have for execution. @@ -884,6 +884,7 @@ def execute_ready_steps(self): # NOTE: verify this actually updates avail_procs on the fly, thus allowing the # available tasks to be fully consumed before going back to sleep + LOGGER.debug("Attempting to submit step {} with total procs = {}, available procs = {}".format(_record.step.name, adapter.total_procs, adapter.avail_procs)) avail_procs = adapter.avail_procs step_procs = _record.step.run.get("procs") if not step_procs: @@ -899,6 +900,7 @@ def execute_ready_steps(self): if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) + # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index ae3b2b542..3286e5cbf 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -30,6 +30,9 @@ """Local interface implementation.""" import logging import os +import psutil +import signal + from concurrent.futures import ThreadPoolExecutor from concurrent.futures._base import PENDING as future_PENDING @@ -75,23 +78,15 @@ def __init__(self, **kwargs): super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys - self.add_batch_parameter("procs", int(kwargs.pop("procs", "1"))) - #self.add_batch_parameter("walltime", kwargs.pop("walltime", ) - # NOTE: add walltime, use this as a timeout on the futures? + self.add_batch_parameter("proc_count", int(kwargs.pop("proc_count", "1"))) self._header = { "procs": "# procs = {procs}", } - # Maybe need better types here for populating the queues? - self.START = 0 - self.CANCEL = 1 - self.DONE = 2 - - self.total_procs = int(kwargs.pop("proc_count", "1")) + self.total_procs = self._batch['proc_count'] self.avail_procs = self.total_procs self.executor = ThreadPoolExecutor(max_workers=self.total_procs) - #self.scheduler_thread = Thread(target=self._scheduler_loop()) def __getstate__(self): """Helper for excluding threadpool from pickling""" @@ -116,7 +111,7 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): for task_id in rm_tasks: if task_id in self.running_steps: self.avail_procs += self.running_steps[task_id][1] # give back resources - + LOGGER.debug("Removing task {} from running_steps".format( self.running_steps.pop(task_id))) else: @@ -290,14 +285,14 @@ def cancel_jobs(self, joblist): # Interrupt running subprocesses try: process, procs = self.running_steps[fut] - process.kill() # better way to kill it? - fut.get_result() # wait on future to exit + self._kill(process.pid) #process.kill() # better way to kill it? + fut.result() # wait on future to exit except KeyError: LOGGER.error("Error, future {}, no longer in running step list".format(fut)) retcode += 1 except: # TODO: catch exceptions from kill(), get_result() - LOGGER.error("Error, unexpected behavior trying to cancel future {}, " + LOGGER.exception("Error, unexpected behavior trying to cancel future {}, " "and subprocess with id {}".format(fut, process.pid)) retcode += 1 @@ -315,6 +310,31 @@ def cancel_jobs(self, joblist): return _record + @staticmethod + def _kill(subprocess_pid, sig=signal.SIGTERM, include_parent=True): + """Kill a process tree (including grandchildren) with signal + "sig" and return a (gone, still_alive) tuple. + "on_terminate", if specified, is a callabck function which is + called as soon as a child terminates. + + NOTE: borrowed from recipe in psutil docs + """ + assert subprocess_pid != os.getpid(), "won't kill myself" + parent = psutil.Process(subprocess_pid) + children = parent.children(recursive=True) + if include_parent: + children.append(parent) + + for p in children: + p.send_signal(sig) + LOGGER.debug("Killing process {}".format(p.name())) + + gone, alive = psutil.wait_procs(children, timeout=5, + callback=None) + + LOGGER.debug(" Gone, alive = {}, {}".format(gone, alive)) + # return (gone, alive) + def _submit(self, p, step, path, cwd, job_map=None, env=None): """ Execute the step locally. From 579d6298d5d657d7f94f1d6fd2137443fb222a24 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 4 May 2020 23:20:56 -0700 Subject: [PATCH 08/27] Update jobid to use uuid, fix handling of future completion and job state updates --- .../datastructures/core/executiongraph.py | 26 ---- .../script/localparscriptadapter.py | 117 ++++++++++-------- 2 files changed, 68 insertions(+), 75 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index a55419712..014d09fc1 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -831,32 +831,6 @@ def execute_ready_steps(self): _available = min(_available, len(self.ready_steps)) LOGGER.info("Found %d available slots...", _available) - - - # print("Num threads: {}".format(nthreads)) - # with futures.ThreadPoolExecutor(max_workers=nthreads) as executor: - # steps_to_do = [] - # for i in range(0, _available): - # # Pop the record and execute using the helper method. - # _record = self.values[self.ready_steps.popleft()] - - # # If we get to this point and we've cancelled, cancel the record. - # if self.is_canceled: - # logger.info("Cancelling '%s' -- continuing.", _record.name) - # _record.mark_end(State.CANCELLED) - # self.cancelled_steps.add(_record.name) - # continue - - # logger.debug("Launching job %d -- %s", i, _record.name) - # steps_to_do.append(executor.submit(self._execute_record, _record, adapter)) - - # for step_future in futures.as_completed(steps_to_do): - # if not step_future: - # print("Encountered None instead of a future.") - # else: - # res = step_future.result() - - # for i in range(0, _available): # # Pop the record and execute using the helper method. # _record = self.values[self.ready_steps.popleft()] diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 3286e5cbf..ee13ac783 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -32,6 +32,7 @@ import os import psutil import signal +import uuid from concurrent.futures import ThreadPoolExecutor from concurrent.futures._base import PENDING as future_PENDING @@ -57,6 +58,7 @@ class LocalParallelScriptAdapter(SchedulerScriptAdapter): executor = None running_steps = {} # needs to be singleton, threadsafe + done_steps = {} # storing completed futures/results to avoid losing them tasks_lock = RLock() total_procs = 1 avail_procs = 1 @@ -110,10 +112,12 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): with self.tasks_lock: for task_id in rm_tasks: if task_id in self.running_steps: - self.avail_procs += self.running_steps[task_id][1] # give back resources + self.avail_procs += self.running_steps[task_id][2] # give back resources - LOGGER.debug("Removing task {} from running_steps".format( - self.running_steps.pop(task_id))) + self.done_steps[task_id] = self.running_steps.pop(task_id) + + LOGGER.debug("Removing task {}, {} from running_steps".format( + task_id, self.done_steps[task_id])) else: LOGGER.error("Tried removing task {} from running_steps, " "but it was not found.".format(task_id)) @@ -225,41 +229,56 @@ def check_jobs(self, joblist): status = {} status_code = JobStatusCode.OK - - for fut in joblist: - LOGGER.debug("Looking for job with Future %s", fut) + LOGGER.debug("Checking jobs {}".format(joblist)) + LOGGER.debug(" Currently running steps: {}".format(self.running_steps)) + for jid in joblist: + LOGGER.debug("Looking for job with id {}".format(jid)) + + if jid in self.running_steps: + LOGGER.debug("Job with id {} is in running steps".format(jid)) + fut, pid, step_procs = self.running_steps[jid] + LOGGER.debug("Job with id {} has future {} with states: done = {}, running = {}, _state = {}".format( + jid, fut, fut.done(), fut.running(), fut._state)) + + elif jid in self.done_steps: + fut, pid, step_procs = self.done_steps.pop(jid) + LOGGER.debug("Job with id {} has future {} with states: done = {}, running = {}, _state = {}".format( + jid, fut, fut.done(), fut.running(), fut._state)) + #status[jid] = State.FINISHED + + else: + LOGGER.debug("Job with id {} is not found in running steps".format(jid)) + # what state reaches this, and what's an appropriate return code? + continue # Future states: running, done, cancelled if fut.done(): # Check result/process retcode: subprocess errors caught here - if fut in self.running_steps: - # NOTE: does the avail_procs in update_running_steps always catch ending or - # is there some other check needed here? - result=fut.result() # this the right place to catch this? - LOGGER.debug("Fut {} has result: {}".format(fut, result)) - status[fut] = State.FINISHING - - else: - status[fut] = State.FINISHED + # if fut in self.running_steps: + # NOTE: does the avail_procs in update_running_steps always catch ending or + # is there some other check needed here? -> save jid and remove from running steps? + result = fut.result() # this the right place to catch this? + LOGGER.debug("Job {}, has result: {}".format(jid, result)) + status[jid] = State.FINISHED elif fut.running(): - status[fut] = State.RUNNING + status[jid] = State.RUNNING elif fut.cancelled(): # What about cancelled pid's? that would show up under fut.done()... - status[fut] = State.CANCELLED + status[jid] = State.CANCELLED elif fut._state == future_PENDING: - LOGGER.debug("Fut {} is pending".format(fut)) - status[fut] = State.PENDING + LOGGER.debug("Job {}, with Fut {} is pending".format(jid, fut)) + status[jid] = State.PENDING else: - print("Fut {} with unknown state".format(fut)) - if fut in self.running_steps: - LOGGER.debug("Fut {} is running?".format(fut)) - LOGGER.debug("Fut {} state: running {}, done {}, cancelled {}".format( - fut, fut.done(), fut.running(), fut.cancelled())) - status[fut] = State.UNKNOWN # this ever reached, and if so how + print("Job {}, with Fut {} with unknown state".format(jid, fut)) + if jid in self.running_steps: + LOGGER.debug("Job {}, with Fut {} is running?".format(jid, fut)) + LOGGER.debug("Job {}, Fut {} state: running {}, done {}, cancelled {}".format( + jid, fut, fut.done(), fut.running(), fut.cancelled())) + status[jid] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR return status_code, status @@ -276,29 +295,33 @@ def cancel_jobs(self, joblist): retcode = 0 - for fut in joblist: - if fut in self.running_steps: + for jid in joblist: + if jid in self.running_steps: + fut, pid, step_procs = self.running_steps[jid] if fut.done(): # cleanup - # Really need _submit to return submission record vs retcode/err? - result = fut.get_result() + # Really need _submit to return submission record vs retcode/err? + result = fut.result() + LOGGER.debug("Removing job {} from running steps. result = {}".format(jid, result)) + else: # Interrupt running subprocesses try: - process, procs = self.running_steps[fut] - self._kill(process.pid) #process.kill() # better way to kill it? - fut.result() # wait on future to exit + self._kill(pid) #process.kill() # better way to kill it? + result=fut.result() # wait on future to exit + LOGGER.debug("Removing job {} from running steps. result = {}".format(jid, result)) except KeyError: LOGGER.error("Error, future {}, no longer in running step list".format(fut)) retcode += 1 except: # TODO: catch exceptions from kill(), get_result() LOGGER.exception("Error, unexpected behavior trying to cancel future {}, " - "and subprocess with id {}".format(fut, process.pid)) + "and subprocess with id {}".format(fut, pid)) retcode += 1 - - if not fut.done() and not fut.running(): + + else: # What state occurs when executiongraph knows of jobs that aren't here? + # if not fut.done() and not fut.running(): # This shouldn't be hit since execution graph only submits if resources available - LOGGER.error("Error, encounterd future {} in unexpected state".format(fut)) + LOGGER.error("Error, encounterd job with id {} in unexpected state".format(jid)) retcode += 1 if retcode == 0: @@ -371,6 +394,7 @@ def _submit(self, p, step, path, cwd, job_map=None, env=None): if retcode == 0: LOGGER.info("Execution returned status OK.") + # REPLACE PID WITH UUID? return SubmissionRecord(SubmissionCode.OK, retcode, pid) else: LOGGER.warning("Execution returned an error: %s", str(err)) @@ -414,30 +438,25 @@ def submit(self, step, path, cwd, job_map=None, env=None): LOGGER.warning("Execution returned an error") _record = SubmissionRecord(SubmissionCode.ERROR, -1) - self.avail_procs += step_procs + #self.avail_procs += step_procs return _record # self.avail_procs -= step._procs # except: - # Update running task list. Thread safe list be better? - - self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) + # Update running task list. + jid = uuid.uuid4() + self._update_running_steps(add_tasks=[{jid: (fut, p.pid, step_procs)}]) # fut.add_done_callback(func_partial(self._task_callback, fut, rm_func=self._update_running_steps)) - fut.add_done_callback(self.wrap_callback(self._update_running_steps)) + fut.add_done_callback(self.wrap_callback(self._update_running_steps, jid)) LOGGER.info("Execution returned status OK.") - return SubmissionRecord(SubmissionCode.OK, 0, fut) + return SubmissionRecord(SubmissionCode.OK, 0, jid) @staticmethod - def wrap_callback(rm_func): + def wrap_callback(rm_func, jid): def task_callback(fut): #fut.result() - rm_func(add_tasks=None, rm_tasks=[fut]) + rm_func(add_tasks=None, rm_tasks=[jid]) return task_callback - - @staticmethod - def _task_callback(fut, rm_func): - fut.get_result() - rm_func(add_tasks=None, rm_tasks=[fut]) From e1eaa80ec953dc6743f3ac67c493408761496af0 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Tue, 5 May 2020 23:28:02 -0700 Subject: [PATCH 09/27] Avoid popping ready steps until sure it will run, add early exit of submit loop --- maestrowf/datastructures/core/executiongraph.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 014d09fc1..eca565df9 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -847,7 +847,7 @@ def execute_ready_steps(self): for i in range(0, _available): # Pop the record and execute using the helper method. - _record = self.values[self.ready_steps.popleft()] + _record = self.values[self.ready_steps[0]] # If we get to this point and we've cancelled, cancel the record. if self.is_canceled: @@ -875,9 +875,13 @@ def execute_ready_steps(self): # NOTE: better place to set this default, and do type conversions? if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) + self.ready_steps.popleft() # remove key now that it's sure to run self._execute_record(_record, adapter) else: LOGGER.debug("step_procs thought > avail_procs: %d : %d", step_procs, avail_procs) + if avail_procs == 0: + break # exit loop early to avoid needless churn + # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() From 2db25108fb64455836957d49152d5a2b3f266f88 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 10 Apr 2020 16:10:16 -0700 Subject: [PATCH 10/27] Check point on threaded local script adapter --- .../datastructures/core/executiongraph.py | 57 +++- maestrowf/datastructures/core/study.py | 1 + .../script/localparscriptadapter.py | 298 ++++++++++++++++++ maestrowf/maestro.py | 3 + 4 files changed, 357 insertions(+), 2 deletions(-) create mode 100644 maestrowf/interfaces/script/localparscriptadapter.py diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index b6395a3f1..c36f17dd8 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -7,6 +7,7 @@ import shutil import tempfile from filelock import FileLock, Timeout +from concurrent import futures from maestrowf.abstracts import PickleInterface from maestrowf.abstracts.enums import JobStatusCode, State, SubmissionCode, \ @@ -338,6 +339,8 @@ def __init__(self, submission_attempts=1, submission_throttle=0, self._submission_throttle = submission_throttle self.dry_run = dry_run + self._local_procs = local_procs + # A map that tracks the dependencies of a step. # NOTE: I don't know how performant the Python dict structure is, but # we'll use it for now. I think this may want to be changed to an AVL @@ -427,6 +430,7 @@ def set_adapter(self, adapter): LOGGER.error(msg) raise TypeError(msg) + print("Adapter set to: {}".format(adapter)) self._adapter = adapter def add_description(self, name, description, **kwargs): @@ -821,6 +825,15 @@ def execute_ready_steps(self): # We now have a collection of ready steps. Execute. # If we don't have a submission limit, go ahead and submit all. + # Check requested resources -> nprocs + nthreads = 1 + if self._local_procs > 0: + nthreads = self._local_procs + + if adapter.total_procs != nthreads: + adapter.total_procs = nthreads + adapter.avail_procs = nthreads + if self._submission_throttle == 0: LOGGER.info("Launching all ready steps...") _available = len(self.ready_steps) @@ -837,6 +850,44 @@ def execute_ready_steps(self): _available = min(_available, len(self.ready_steps)) LOGGER.info("Found %d available slots...", _available) + # print("Num threads: {}".format(nthreads)) + # with futures.ThreadPoolExecutor(max_workers=nthreads) as executor: + # steps_to_do = [] + # for i in range(0, _available): + # # Pop the record and execute using the helper method. + # _record = self.values[self.ready_steps.popleft()] + + # # If we get to this point and we've cancelled, cancel the record. + # if self.is_canceled: + # logger.info("Cancelling '%s' -- continuing.", _record.name) + # _record.mark_end(State.CANCELLED) + # self.cancelled_steps.add(_record.name) + # continue + + # logger.debug("Launching job %d -- %s", i, _record.name) + # steps_to_do.append(executor.submit(self._execute_record, _record, adapter)) + + # for step_future in futures.as_completed(steps_to_do): + # if not step_future: + # print("Encountered None instead of a future.") + # else: + # res = step_future.result() + + + # for i in range(0, _available): + # # Pop the record and execute using the helper method. + # _record = self.values[self.ready_steps.popleft()] + + # # If we get to this point and we've cancelled, cancel the record. + # if self.is_canceled: + # logger.info("Cancelling '%s' -- continuing.", _record.name) + # _record.mark_end(State.CANCELLED) + # self.cancelled_steps.add(_record.name) + # continue + + # logger.debug("Launching job %d -- %s", i, _record.name) + # self._execute_record(_record, adapter) + for i in range(0, _available): # Pop the record and execute using the helper method. _record = self.values[self.ready_steps.popleft()] @@ -848,8 +899,10 @@ def execute_ready_steps(self): self.cancelled_steps.add(_record.name) continue - LOGGER.debug("Launching job %d -- %s", i, _record.name) - self._execute_record(_record, adapter) + avail_procs = adapter.avail_procs + if _record.step._procs <= avail_procs: + logger.debug("Launching job %d -- %s", i, _record.name) + self._execute_record(_record, adapter) # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/datastructures/core/study.py b/maestrowf/datastructures/core/study.py index e6de13a73..58908b005 100644 --- a/maestrowf/datastructures/core/study.py +++ b/maestrowf/datastructures/core/study.py @@ -416,6 +416,7 @@ def configure_study(self, submission_attempts=1, restart_limit=1, self._submission_attempts = submission_attempts self._restart_limit = restart_limit self._submission_throttle = throttle + self._local_procs = local_procs self._use_tmp = use_tmp self._hash_ws = hash_ws self._dry_run = dry_run diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py new file mode 100644 index 000000000..6decf85c2 --- /dev/null +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -0,0 +1,298 @@ +############################################################################### +# Copyright (c) 2017, Lawrence Livermore National Security, LLC. +# Produced at the Lawrence Livermore National Laboratory +# Written by Francesco Di Natale, dinatale3@llnl.gov. +# +# LLNL-CODE-734340 +# All rights reserved. +# This file is part of MaestroWF, Version: 1.0.0. +# +# For details, see https://github.com/LLNL/maestrowf. +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +############################################################################### + +"""Local interface implementation.""" +import logging +import os +from concurrent.futures import ThreadPoolExecutor +from queue import Queue # priority queue too? +from threading import Thread + +from maestrowf.abstracts.enums import JobStatusCode, SubmissionCode, \ + CancelCode +from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord +from maestrowf.abstracts.interfaces import ScriptAdapter +from maestrowf.abstracts import Singleton +from maestrowf.utils import start_process + +LOGGER = logging.getLogger(__name__) + + +class LocalParallelScriptAdapter(ScriptAdapter): + """A ScriptAdapter class for interfacing for parallel local execution.""" + + key = "local_parallel" + + executor = None + scheduler_thread = None + running_steps = {} # needs to be singleton, threadsafe + submit_queue = None + total_procs = 1 + avail_procs = 1 + + def __init__(self, **kwargs): + """ + Initialize an instance of the LocalParallelScriptAdapter. + + The LocalParallelScriptAdapter is the adapter that is used for workflows that + will execute on the user's machine. This adapter constructs shell scripts for + a StudyStep based on user set defaults and local settings present in each step. + + These scripts are submitted to a local scheduler for asynchronous execution + and monitoring. + + :param **kwargs: A dictionary with default settings for the adapter. + """ + LOGGER.debug("kwargs\n--------------------------\n%s", kwargs) + super(LocalParallelScriptAdapter, self).__init__(**kwargs) + + # Register keys + self.add_batch_parameter("procs", kwargs.pop("procs", "1")) + self.add_batch_parameter("walltime") + + self._header = { + "procs": "# procs = {procs}", + } + + # Maybe need better types here for populating the queues? + self.START = 0 + self.CANCEL = 1 + self.DONE = 2 + + self.total_procs = kwargs.pop("proc_count", "1") + self.avail_procs = self.total_procs + self.submit_queue = Queue() + self.submit_queue.put(self.START) + self.executor = ThreadPoolExecutor(max_workers=self.total_procs) + self.scheduler_thread = Thread(target=self._scheduler_loop()) + + + def _write_script(self, ws_path, step): + """ + Write a shell script to the workspace of a workflow step. + + The job_map optional parameter is a map of workflow step names to job + identifiers. This parameter so far is only planned to be used when a + study is configured to be launched in one go (more or less a script + chain using a scheduler's dependency setting). The functionality of + the parameter may change depending on both future intended use. + + :param ws_path: Path to the workspace directory of the step. + :param step: An instance of a StudyStep. + :returns: False (will not be scheduled), the path to the + written script for run["cmd"], and the path to the script written + for run["restart"] (if it exists). + """ + cmd = step.run["cmd"] + restart = step.run["restart"] + to_be_scheduled = False + + fname = "{}.sh".format(step.name) + script_path = os.path.join(ws_path, fname) + with open(script_path, "w") as script: + script.write("#!{0}\n\n{1}\n".format(self._exec, cmd)) + + if restart: + rname = "{}.restart.sh".format(step.name) + restart_path = os.path.join(ws_path, rname) + + with open(restart_path, "w") as script: + script.write("#!{0}\n\n{1}\n".format(self._exec, restart)) + else: + restart_path = None + + return to_be_scheduled, script_path, restart_path + + def check_jobs(self, joblist): + """ + For the given job list, query execution status. + + This method uses the scontrol show job command and does a + regex search for job information. + + :param joblist: A list of job identifiers to be queried. + :returns: The return code of the status query, and a dictionary of job + identifiers to their status. + """ + + status = {} + status_code = JobStatusCode.OK + + for fut in joblist: # Note, + LOGGER.debug("Looking for job with Future id %s", fut.id) + + # Future states: running, done, cancelled + if fut.done(): + # Check result/process retcode: subprocess errors caught here + if fut in self.running_steps: + result=fut.get_result() # this the right place to catch this? + + status[fut.id] = State.FINISHING + + else: + status[fut.id] = State.FINISHED + + elif fut.running(): + status[fut.id] = State.RUNNING + + elif fut.cancelled(): + # What about cancelled pid's? that would show up under fut.done()... + status[fut.id] = State.CANCELLED + + else: + status[fut.id] = State.UNKNOWN # this ever reached, and if so how + status_code = JobStatusCode.ERROR + + return status_code, status + + def cancel_jobs(self, joblist): + """ + For the given job list, cancel each job. + + :param joblist: A list of job identifiers to be cancelled. + :returns: The return code to indicate if jobs were cancelled. + """ + if not joblist: + return CancellationRecord(CancelCode.OK, 0) + + retcode = 0 + + for fut in joblist: + if fut in self.running_steps: + if fut.done(): # cleanup + # Really need _submit to return submission record vs retcode/err? + result = fut.get_result() + else: + # Interrupt running subprocesses + try: + process, procs = self.running_steps[fut] + process.kill() # better way to kill it? + fut.get_result() # wait on future to exit + except KeyError: + LOGGER.error("Error, future {}, no longer in running step list".format(fut)) + retcode += 1 + + except: # TODO: catch exceptions from kill(), get_result() + LOGGER.error("Error, unexpected behavior trying to cancel future {}, " + "and subprocess with id {}".format(fut, process.pid)) + retcode += 1 + + if not fut.done() and not fut.running(): + # This shouldn't be hit since execution graph only submits if resources available + LOGGER.error("Error, encounterd future {} in unexpected state".format(fut)) + retcode += 1 + + if retcode == 0: + _record = CancellationRecord(CancelCode.OK, retcode) + else: + LOGGER.error("Error code '%s' seen. Unexpected behavior " + "encountered.") # NOTE: what does logger.error inject in %s? + _record = CancellationRecord(CancelCode.ERROR, retcode) + + return _record + + def _submit(self, p, step, path, cwd, job_map=None, env=None): + """ + Execute the step locally. + + If cwd is specified, the submit method will operate outside of the path + specified by the 'cwd' parameter. + If env is specified, the submit method will set the environment + variables for submission to the specified values. The 'env' parameter + should be a dictionary of environment variables. + + :param p: Subprocess object + :param step: An instance of a StudyStep. + :param path: Path to the script to be executed. + :param cwd: Path to the current working directory. + :param job_map: A map of workflow step names to their job identifiers. + :param env: A dict containing a modified environment for execution. + :returns: The return code of the submission command and job identiifer. + """ + LOGGER.debug("cwd = %s", cwd) + LOGGER.debug("Script to execute: %s", path) + + pid = p.pid + output, err = p.communicate() + retcode = p.wait() + + o_path = os.path.join(cwd, "{}.out".format(step.name)) + e_path = os.path.join(cwd, "{}.err".format(step.name)) + + with open(o_path, "w") as out: + out.write(output) + + with open(e_path, "w") as out: + out.write(err) + + if retcode == 0: + LOGGER.info("Execution returned status OK.") + return SubmissionRecord(SubmissionCode.OK, retcode, pid) + else: + LOGGER.warning("Execution returned an error: %s", str(err)) + _record = SubmissionRecord(SubmissionCode.ERROR, retcode, pid) + _record.add_info("stderr", str(err)) + return _record + + def submit(self, step, path, cwd, job_map=None, env=None): + """ + Execute the step locally. + + If cwd is specified, the submit method will operate outside of the path + specified by the 'cwd' parameter. + If env is specified, the submit method will set the environment + variables for submission to the specified values. The 'env' parameter + should be a dictionary of environment variables. + + :param step: An instance of a StudyStep. + :param path: Path to the script to be executed. + :param cwd: Path to the current working directory. + :param job_map: A map of workflow step names to their job identifiers. + :param env: A dict containing a modified environment for execution. + :returns: The return code of the submission command and job identiifer. + """ + try: + p = start_process(path, shell=False, cwd=cwd, env=env) + fut = executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + + except: # add some relevant exceptions here.. + LOGGER.warning("Execution returned an error") + + _record = SubmissionRecord(SubmissionCode.ERROR, -1) + self.avail_procs += step._procs + return _record + + # self.avail_procs -= step._procs + # except: + + # Update running task list. Thread safe list be better? + self.running_steps[fut] = (p, step._procs) + + LOGGER.info("Execution returned status OK.") + return SubmissionRecord(SubmissionCode.OK, 0, fut) diff --git a/maestrowf/maestro.py b/maestrowf/maestro.py index 0daf23a79..688463dbc 100644 --- a/maestrowf/maestro.py +++ b/maestrowf/maestro.py @@ -328,6 +328,9 @@ def setup_argparser(): help="Maximum number of inflight jobs allowed to execute " "simultaneously (0 denotes not throttling)." "[Default: %(default)d]") + run.add_argument("-lp", "--local_procs", type=int, default=1, + help="Max number of processors to use simultaneously." + "[Default: %(default)d]") run.add_argument("-s", "--sleeptime", type=int, default=60, help="Amount of time (in seconds) for the manager to " "wait between job status checks. [Default: %(default)d]") From 492dc111dc28b7c93bce22e4347ca398d69561dd Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 10 Apr 2020 19:36:35 -0700 Subject: [PATCH 11/27] Fixes and hacks to get demo spec running. --- .../datastructures/core/executiongraph.py | 16 ++- maestrowf/interfaces/__init__.py | 1 + .../script/localparscriptadapter.py | 123 +++++++++++++++--- 3 files changed, 120 insertions(+), 20 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index c36f17dd8..4526bc736 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -428,6 +428,7 @@ def set_adapter(self, adapter): msg = "'{}' adapter must be specfied in ScriptAdapterFactory." \ .format(adapter) LOGGER.error(msg) + LOGGER.error("Valid adapters: {}".format(ScriptAdapterFactory.get_valid_adapters())) raise TypeError(msg) print("Adapter set to: {}".format(adapter)) @@ -900,8 +901,19 @@ def execute_ready_steps(self): continue avail_procs = adapter.avail_procs - if _record.step._procs <= avail_procs: - logger.debug("Launching job %d -- %s", i, _record.name) + step_procs = _record.step.run.get("procs") + if not step_procs: + step_procs = 1 + else: + try: + step_procs = int(step_procs) + except ValueError: + step_procs = 1 + LOGGER.error("Setting step {} with no 'procs' attribute to use 1 processor.".format(_record.step.name)) + + # NOTE: better place to set this default, and do type conversions? + if step_procs <= avail_procs: + LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) # check the status of the study upon finishing this round of execution diff --git a/maestrowf/interfaces/__init__.py b/maestrowf/interfaces/__init__.py index 156760a6f..671e2b1b8 100644 --- a/maestrowf/interfaces/__init__.py +++ b/maestrowf/interfaces/__init__.py @@ -52,6 +52,7 @@ def iter_adapters(): mods = [(name, ispkg) for finder, name, ispkg in pkgutil.iter_modules( loader.load_module('maestrowf.interfaces.script').__path__, loader.load_module('maestrowf.interfaces.script').__name__ + ".")] + cs = [] for name, _ in mods: # get loader for every module diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 6decf85c2..600e99f37 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -31,28 +31,28 @@ import logging import os from concurrent.futures import ThreadPoolExecutor -from queue import Queue # priority queue too? -from threading import Thread +from functools import partial as func_partial +from threading import Thread, RLock from maestrowf.abstracts.enums import JobStatusCode, SubmissionCode, \ - CancelCode + CancelCode, State from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord -from maestrowf.abstracts.interfaces import ScriptAdapter +from maestrowf.abstracts.interfaces import SchedulerScriptAdapter from maestrowf.abstracts import Singleton from maestrowf.utils import start_process + LOGGER = logging.getLogger(__name__) -class LocalParallelScriptAdapter(ScriptAdapter): +class LocalParallelScriptAdapter(SchedulerScriptAdapter): """A ScriptAdapter class for interfacing for parallel local execution.""" key = "local_parallel" executor = None - scheduler_thread = None running_steps = {} # needs to be singleton, threadsafe - submit_queue = None + tasks_lock = RLock() total_procs = 1 avail_procs = 1 @@ -73,8 +73,9 @@ def __init__(self, **kwargs): super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys - self.add_batch_parameter("procs", kwargs.pop("procs", "1")) - self.add_batch_parameter("walltime") + self.add_batch_parameter("procs", int(kwargs.pop("procs", "1"))) + #self.add_batch_parameter("walltime", kwargs.pop("walltime", ) + # NOTE: add walltime, use this as a timeout on the futures? self._header = { "procs": "# procs = {procs}", @@ -85,14 +86,83 @@ def __init__(self, **kwargs): self.CANCEL = 1 self.DONE = 2 - self.total_procs = kwargs.pop("proc_count", "1") + self.total_procs = int(kwargs.pop("proc_count", "1")) self.avail_procs = self.total_procs - self.submit_queue = Queue() - self.submit_queue.put(self.START) self.executor = ThreadPoolExecutor(max_workers=self.total_procs) - self.scheduler_thread = Thread(target=self._scheduler_loop()) + #self.scheduler_thread = Thread(target=self._scheduler_loop()) + + def _update_running_steps(self, add_tasks=None, rm_tasks=None): + """Thread safe addition/removal of tasks from running_steps dict""" + if not add_tasks: + add_tasks = [] + + if not rm_tasks: + rm_tasks = [] + + with self.tasks_lock: + for task_id in rm_tasks: + if task_id in self.running_steps: + LOGGER.debug("Removing task {} from running_steps".format( + self.running_steps.pop(task_id))) + else: + LOGGER.error("Tried removing task {} from running_steps, " + "but it was not found.".format(task_id)) + + for task in add_tasks: + for task_id in task: + if task_id in self.running_steps: + LOGGER.error("Tried adding already tracked task {} " + "to running_steps.".format(task_id)) + else: + LOGGER.debug("Adding task {} to running_steps".format(task_id)) + self.running_steps[task_id] = task[task_id] + + # Want any status codes returned here if errors triggered? + + def get_header(self, step): + """ + Generate the header present at the top of execution scripts. + + :param step: A StudyStep instance. + :returns: A string of the header based on internal batch parameters and + the parameter step. + """ + return "" + + def get_parallelize_command(self, procs, nodes, **kwargs): + """ + Generate the parallelization segement of the command line. + + :param procs: Number of processors to allocate to the parallel call. + :param nodes: Number of nodes to allocate to the parallel call + (default = 1). + :returns: A string of the parallelize command configured using nodes + and procs. + :NOTE: this is currently a dummy method -> rework to add user specified replacements later + """ + return "" + + def _state(self, fut): + """ + Map a scheduler specific job state to a Study.State enum. + + :param fut: Future instance representing a task + :returns: A Study.State enum corresponding to parameter job_state. + """ + # NOTE: should all of this replace what's in check_jobs -> what about error code there? + if fut.running(): + return State.RUNNING + elif fut.done(): + if fut in self.running_steps: + return State.FINISHING + else: + return State.FINISHED + elif fut.cancelled(): + return State.CANCELLED + + else: + return State.UNKNOWN - def _write_script(self, ws_path, step): """ Write a shell script to the workspace of a workflow step. @@ -109,9 +179,10 @@ def _write_script(self, ws_path, step): written script for run["cmd"], and the path to the script written for run["restart"] (if it exists). """ + # THIS IS A HACK FOR NOW: make better use of get_scheduler_command later cmd = step.run["cmd"] restart = step.run["restart"] - to_be_scheduled = False + to_be_scheduled, _, _ = self.get_scheduler_command(step) fname = "{}.sh".format(step.name) script_path = os.path.join(ws_path, fname) @@ -279,20 +350,36 @@ def submit(self, step, path, cwd, job_map=None, env=None): """ try: p = start_process(path, shell=False, cwd=cwd, env=env) - fut = executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + fut = self.executor.submit(self._submit, p, step, path, cwd, job_map=None, env=None) + try: + step_procs = step.run.get("procs") + if not step_procs: + step_procs = 1 + else: + step_procs = int(step_procs) + except ValueError: + step_procs = 1 + LOGGER.error("Starting step {} with no 'procs' attribute with 1 processor.".format(_record.step.name)) + + self.avail_procs -= step_procs except: # add some relevant exceptions here.. LOGGER.warning("Execution returned an error") _record = SubmissionRecord(SubmissionCode.ERROR, -1) - self.avail_procs += step._procs + self.avail_procs += step_procs return _record # self.avail_procs -= step._procs # except: # Update running task list. Thread safe list be better? - self.running_steps[fut] = (p, step._procs) + + self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) + fut.add_done_callback(func_partial(self._task_callback, rm_func=self._update_running_steps)) LOGGER.info("Execution returned status OK.") return SubmissionRecord(SubmissionCode.OK, 0, fut) + + def _task_callback(fut, rm_func): + rm_func(add_tasks=None, rm_tasks=[fut]) From c1210103ca8e8f468d78543512429a71e2c2134d Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 13 Apr 2020 18:26:36 -0700 Subject: [PATCH 12/27] Update future callbacks, exclude threadpool from pickling --- .../script/localparscriptadapter.py | 44 +++++++++++++++---- 1 file changed, 35 insertions(+), 9 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 600e99f37..1ba596539 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -91,6 +91,17 @@ def __init__(self, **kwargs): self.executor = ThreadPoolExecutor(max_workers=self.total_procs) #self.scheduler_thread = Thread(target=self._scheduler_loop()) + def __getstate__(self): + """Helper for excluding threadpool from pickling""" + state = self.__dict__.copy() + del state["executor"] + return state + + def __setstate__(self, state): + self.__dict__.update(state) + # Add baz back since it doesn't exist in the pickle + self.executor = ThreadPoolExecutor(max_workers=self.total_procs) + def _update_running_steps(self, add_tasks=None, rm_tasks=None): """Thread safe addition/removal of tasks from running_steps dict""" if not add_tasks: @@ -102,6 +113,8 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): with self.tasks_lock: for task_id in rm_tasks: if task_id in self.running_steps: + self.avail_procs += self.running_steps[task_id][1] # give back resources + LOGGER.debug("Removing task {} from running_steps".format( self.running_steps.pop(task_id))) else: @@ -215,29 +228,31 @@ def check_jobs(self, joblist): status = {} status_code = JobStatusCode.OK - for fut in joblist: # Note, - LOGGER.debug("Looking for job with Future id %s", fut.id) + for fut in joblist: + LOGGER.debug("Looking for job with Future %s", fut) # Future states: running, done, cancelled if fut.done(): # Check result/process retcode: subprocess errors caught here if fut in self.running_steps: - result=fut.get_result() # this the right place to catch this? + # NOTE: does the avail_procs in update_running_steps always catch ending or + # is there some other check needed here? + result=fut.result() # this the right place to catch this? - status[fut.id] = State.FINISHING + status[fut] = State.FINISHING else: - status[fut.id] = State.FINISHED + status[fut] = State.FINISHED elif fut.running(): - status[fut.id] = State.RUNNING + status[fut] = State.RUNNING elif fut.cancelled(): # What about cancelled pid's? that would show up under fut.done()... - status[fut.id] = State.CANCELLED + status[fut] = State.CANCELLED else: - status[fut.id] = State.UNKNOWN # this ever reached, and if so how + status[fut] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR return status_code, status @@ -376,10 +391,21 @@ def submit(self, step, path, cwd, job_map=None, env=None): # Update running task list. Thread safe list be better? self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) - fut.add_done_callback(func_partial(self._task_callback, rm_func=self._update_running_steps)) + # fut.add_done_callback(func_partial(self._task_callback, fut, rm_func=self._update_running_steps)) + fut.add_done_callback(self.wrap_callback(self._update_running_steps)) LOGGER.info("Execution returned status OK.") return SubmissionRecord(SubmissionCode.OK, 0, fut) + @staticmethod + def wrap_callback(rm_func): + def task_callback(fut): + fut.result() + rm_func(add_tasks=None, rm_tasks=[fut]) + + return task_callback + + @staticmethod def _task_callback(fut, rm_func): + fut.get_result() rm_func(add_tasks=None, rm_tasks=[fut]) From 9232b7bced7cfaf0d0edf082a28ede223a66ca62 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Wed, 15 Apr 2020 00:15:21 -0700 Subject: [PATCH 13/27] Some dirty hacks to deal with hidden pending state. --- .../interfaces/script/localparscriptadapter.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 1ba596539..ae3b2b542 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -31,6 +31,8 @@ import logging import os from concurrent.futures import ThreadPoolExecutor +from concurrent.futures._base import PENDING as future_PENDING + from functools import partial as func_partial from threading import Thread, RLock @@ -172,7 +174,8 @@ def _state(self, fut): return State.FINISHED elif fut.cancelled(): return State.CANCELLED - + elif fut._state == future_PENDING: + return State.PENDING else: return State.UNKNOWN @@ -238,7 +241,7 @@ def check_jobs(self, joblist): # NOTE: does the avail_procs in update_running_steps always catch ending or # is there some other check needed here? result=fut.result() # this the right place to catch this? - + LOGGER.debug("Fut {} has result: {}".format(fut, result)) status[fut] = State.FINISHING else: @@ -251,7 +254,16 @@ def check_jobs(self, joblist): # What about cancelled pid's? that would show up under fut.done()... status[fut] = State.CANCELLED + elif fut._state == future_PENDING: + LOGGER.debug("Fut {} is pending".format(fut)) + status[fut] = State.PENDING + else: + print("Fut {} with unknown state".format(fut)) + if fut in self.running_steps: + LOGGER.debug("Fut {} is running?".format(fut)) + LOGGER.debug("Fut {} state: running {}, done {}, cancelled {}".format( + fut, fut.done(), fut.running(), fut.cancelled())) status[fut] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR @@ -400,7 +412,7 @@ def submit(self, step, path, cwd, job_map=None, env=None): @staticmethod def wrap_callback(rm_func): def task_callback(fut): - fut.result() + #fut.result() rm_func(add_tasks=None, rm_tasks=[fut]) return task_callback From 2d32a5140c1c072ddd042f3db9b051d9a6c00a0d Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 20 Apr 2020 14:36:43 -0700 Subject: [PATCH 14/27] Remove unneeded hooks for command line local_procs parameter --- maestrowf/datastructures/core/executiongraph.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 4526bc736..eb47c1471 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -827,13 +827,13 @@ def execute_ready_steps(self): # We now have a collection of ready steps. Execute. # If we don't have a submission limit, go ahead and submit all. # Check requested resources -> nprocs - nthreads = 1 - if self._local_procs > 0: - nthreads = self._local_procs + # nthreads = 1 + # if self._local_procs > 0: + # nthreads = self._local_procs - if adapter.total_procs != nthreads: - adapter.total_procs = nthreads - adapter.avail_procs = nthreads + # if adapter.total_procs != nthreads: + # adapter.total_procs = nthreads + # adapter.avail_procs = nthreads if self._submission_throttle == 0: LOGGER.info("Launching all ready steps...") @@ -900,6 +900,8 @@ def execute_ready_steps(self): self.cancelled_steps.add(_record.name) continue + # NOTE: verify this actually updates avail_procs on the fly, thus allowing the + # available tasks to be fully consumed before going back to sleep avail_procs = adapter.avail_procs step_procs = _record.step.run.get("procs") if not step_procs: From 1383c37df7071fac81cd70d19f8298094b38bd9c Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 24 Apr 2020 11:46:11 -0700 Subject: [PATCH 15/27] Enable cancellation of processes, fix cancel status marking --- .../datastructures/core/executiongraph.py | 4 +- .../script/localparscriptadapter.py | 48 +++++++++++++------ 2 files changed, 37 insertions(+), 15 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index eb47c1471..4688b098c 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -431,7 +431,6 @@ def set_adapter(self, adapter): LOGGER.error("Valid adapters: {}".format(ScriptAdapterFactory.get_valid_adapters())) raise TypeError(msg) - print("Adapter set to: {}".format(adapter)) self._adapter = adapter def add_description(self, name, description, **kwargs): @@ -838,6 +837,7 @@ def execute_ready_steps(self): if self._submission_throttle == 0: LOGGER.info("Launching all ready steps...") _available = len(self.ready_steps) + LOGGER.info("Ready steps: {}".format(self.ready_steps)) # Else, we have a limit -- adhere to it. else: # Compute the number of available slots we have for execution. @@ -902,6 +902,7 @@ def execute_ready_steps(self): # NOTE: verify this actually updates avail_procs on the fly, thus allowing the # available tasks to be fully consumed before going back to sleep + LOGGER.debug("Attempting to submit step {} with total procs = {}, available procs = {}".format(_record.step.name, adapter.total_procs, adapter.avail_procs)) avail_procs = adapter.avail_procs step_procs = _record.step.run.get("procs") if not step_procs: @@ -917,6 +918,7 @@ def execute_ready_steps(self): if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) + # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index ae3b2b542..3286e5cbf 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -30,6 +30,9 @@ """Local interface implementation.""" import logging import os +import psutil +import signal + from concurrent.futures import ThreadPoolExecutor from concurrent.futures._base import PENDING as future_PENDING @@ -75,23 +78,15 @@ def __init__(self, **kwargs): super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys - self.add_batch_parameter("procs", int(kwargs.pop("procs", "1"))) - #self.add_batch_parameter("walltime", kwargs.pop("walltime", ) - # NOTE: add walltime, use this as a timeout on the futures? + self.add_batch_parameter("proc_count", int(kwargs.pop("proc_count", "1"))) self._header = { "procs": "# procs = {procs}", } - # Maybe need better types here for populating the queues? - self.START = 0 - self.CANCEL = 1 - self.DONE = 2 - - self.total_procs = int(kwargs.pop("proc_count", "1")) + self.total_procs = self._batch['proc_count'] self.avail_procs = self.total_procs self.executor = ThreadPoolExecutor(max_workers=self.total_procs) - #self.scheduler_thread = Thread(target=self._scheduler_loop()) def __getstate__(self): """Helper for excluding threadpool from pickling""" @@ -116,7 +111,7 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): for task_id in rm_tasks: if task_id in self.running_steps: self.avail_procs += self.running_steps[task_id][1] # give back resources - + LOGGER.debug("Removing task {} from running_steps".format( self.running_steps.pop(task_id))) else: @@ -290,14 +285,14 @@ def cancel_jobs(self, joblist): # Interrupt running subprocesses try: process, procs = self.running_steps[fut] - process.kill() # better way to kill it? - fut.get_result() # wait on future to exit + self._kill(process.pid) #process.kill() # better way to kill it? + fut.result() # wait on future to exit except KeyError: LOGGER.error("Error, future {}, no longer in running step list".format(fut)) retcode += 1 except: # TODO: catch exceptions from kill(), get_result() - LOGGER.error("Error, unexpected behavior trying to cancel future {}, " + LOGGER.exception("Error, unexpected behavior trying to cancel future {}, " "and subprocess with id {}".format(fut, process.pid)) retcode += 1 @@ -315,6 +310,31 @@ def cancel_jobs(self, joblist): return _record + @staticmethod + def _kill(subprocess_pid, sig=signal.SIGTERM, include_parent=True): + """Kill a process tree (including grandchildren) with signal + "sig" and return a (gone, still_alive) tuple. + "on_terminate", if specified, is a callabck function which is + called as soon as a child terminates. + + NOTE: borrowed from recipe in psutil docs + """ + assert subprocess_pid != os.getpid(), "won't kill myself" + parent = psutil.Process(subprocess_pid) + children = parent.children(recursive=True) + if include_parent: + children.append(parent) + + for p in children: + p.send_signal(sig) + LOGGER.debug("Killing process {}".format(p.name())) + + gone, alive = psutil.wait_procs(children, timeout=5, + callback=None) + + LOGGER.debug(" Gone, alive = {}, {}".format(gone, alive)) + # return (gone, alive) + def _submit(self, p, step, path, cwd, job_map=None, env=None): """ Execute the step locally. From 83739e29693e2936d304ccc854c208fb5856e432 Mon Sep 17 00:00:00 2001 From: Jeff Greenough Date: Fri, 24 Apr 2020 08:47:34 -0700 Subject: [PATCH 16/27] correctly set avail_procs and total_procs for local parallel adapter --- maestrowf/datastructures/core/executiongraph.py | 5 ++++- maestrowf/interfaces/script/localparscriptadapter.py | 9 +++++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 4688b098c..934251751 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -904,6 +904,8 @@ def execute_ready_steps(self): # available tasks to be fully consumed before going back to sleep LOGGER.debug("Attempting to submit step {} with total procs = {}, available procs = {}".format(_record.step.name, adapter.total_procs, adapter.avail_procs)) avail_procs = adapter.avail_procs + LOGGER.debug("avail_procs from the adapter = %d", avail_procs) + LOGGER.debug("total_procs from the adapter = %d", adapter.total_procs) step_procs = _record.step.run.get("procs") if not step_procs: step_procs = 1 @@ -918,7 +920,8 @@ def execute_ready_steps(self): if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) self._execute_record(_record, adapter) - + else: + LOGGER.debug("step_procs thought > avail_procs: %d : %d", step_procs, avail_procs) # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 3286e5cbf..c2167d150 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -84,8 +84,13 @@ def __init__(self, **kwargs): "procs": "# procs = {procs}", } - self.total_procs = self._batch['proc_count'] - self.avail_procs = self.total_procs + # Maybe need better types here for populating the queues? + self.START = 0 + self.CANCEL = 1 + self.DONE = 2 + + self.total_procs = int(kwargs.pop("proc_count", "1")) + self.avail_procs = int(kwargs.pop("avail_procs", "1")) self.executor = ThreadPoolExecutor(max_workers=self.total_procs) def __getstate__(self): From 7ba9ed3ee030887e11da31061866848d29164869 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 4 May 2020 23:20:56 -0700 Subject: [PATCH 17/27] Update jobid to use uuid, fix handling of future completion and job state updates --- .../datastructures/core/executiongraph.py | 24 ---- .../script/localparscriptadapter.py | 117 ++++++++++-------- 2 files changed, 68 insertions(+), 73 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 934251751..a3b323624 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -851,30 +851,6 @@ def execute_ready_steps(self): _available = min(_available, len(self.ready_steps)) LOGGER.info("Found %d available slots...", _available) - # print("Num threads: {}".format(nthreads)) - # with futures.ThreadPoolExecutor(max_workers=nthreads) as executor: - # steps_to_do = [] - # for i in range(0, _available): - # # Pop the record and execute using the helper method. - # _record = self.values[self.ready_steps.popleft()] - - # # If we get to this point and we've cancelled, cancel the record. - # if self.is_canceled: - # logger.info("Cancelling '%s' -- continuing.", _record.name) - # _record.mark_end(State.CANCELLED) - # self.cancelled_steps.add(_record.name) - # continue - - # logger.debug("Launching job %d -- %s", i, _record.name) - # steps_to_do.append(executor.submit(self._execute_record, _record, adapter)) - - # for step_future in futures.as_completed(steps_to_do): - # if not step_future: - # print("Encountered None instead of a future.") - # else: - # res = step_future.result() - - # for i in range(0, _available): # # Pop the record and execute using the helper method. # _record = self.values[self.ready_steps.popleft()] diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index c2167d150..df947280a 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -32,6 +32,7 @@ import os import psutil import signal +import uuid from concurrent.futures import ThreadPoolExecutor from concurrent.futures._base import PENDING as future_PENDING @@ -57,6 +58,7 @@ class LocalParallelScriptAdapter(SchedulerScriptAdapter): executor = None running_steps = {} # needs to be singleton, threadsafe + done_steps = {} # storing completed futures/results to avoid losing them tasks_lock = RLock() total_procs = 1 avail_procs = 1 @@ -115,10 +117,12 @@ def _update_running_steps(self, add_tasks=None, rm_tasks=None): with self.tasks_lock: for task_id in rm_tasks: if task_id in self.running_steps: - self.avail_procs += self.running_steps[task_id][1] # give back resources + self.avail_procs += self.running_steps[task_id][2] # give back resources - LOGGER.debug("Removing task {} from running_steps".format( - self.running_steps.pop(task_id))) + self.done_steps[task_id] = self.running_steps.pop(task_id) + + LOGGER.debug("Removing task {}, {} from running_steps".format( + task_id, self.done_steps[task_id])) else: LOGGER.error("Tried removing task {} from running_steps, " "but it was not found.".format(task_id)) @@ -230,41 +234,56 @@ def check_jobs(self, joblist): status = {} status_code = JobStatusCode.OK - - for fut in joblist: - LOGGER.debug("Looking for job with Future %s", fut) + LOGGER.debug("Checking jobs {}".format(joblist)) + LOGGER.debug(" Currently running steps: {}".format(self.running_steps)) + for jid in joblist: + LOGGER.debug("Looking for job with id {}".format(jid)) + + if jid in self.running_steps: + LOGGER.debug("Job with id {} is in running steps".format(jid)) + fut, pid, step_procs = self.running_steps[jid] + LOGGER.debug("Job with id {} has future {} with states: done = {}, running = {}, _state = {}".format( + jid, fut, fut.done(), fut.running(), fut._state)) + + elif jid in self.done_steps: + fut, pid, step_procs = self.done_steps.pop(jid) + LOGGER.debug("Job with id {} has future {} with states: done = {}, running = {}, _state = {}".format( + jid, fut, fut.done(), fut.running(), fut._state)) + #status[jid] = State.FINISHED + + else: + LOGGER.debug("Job with id {} is not found in running steps".format(jid)) + # what state reaches this, and what's an appropriate return code? + continue # Future states: running, done, cancelled if fut.done(): # Check result/process retcode: subprocess errors caught here - if fut in self.running_steps: - # NOTE: does the avail_procs in update_running_steps always catch ending or - # is there some other check needed here? - result=fut.result() # this the right place to catch this? - LOGGER.debug("Fut {} has result: {}".format(fut, result)) - status[fut] = State.FINISHING - - else: - status[fut] = State.FINISHED + # if fut in self.running_steps: + # NOTE: does the avail_procs in update_running_steps always catch ending or + # is there some other check needed here? -> save jid and remove from running steps? + result = fut.result() # this the right place to catch this? + LOGGER.debug("Job {}, has result: {}".format(jid, result)) + status[jid] = State.FINISHED elif fut.running(): - status[fut] = State.RUNNING + status[jid] = State.RUNNING elif fut.cancelled(): # What about cancelled pid's? that would show up under fut.done()... - status[fut] = State.CANCELLED + status[jid] = State.CANCELLED elif fut._state == future_PENDING: - LOGGER.debug("Fut {} is pending".format(fut)) - status[fut] = State.PENDING + LOGGER.debug("Job {}, with Fut {} is pending".format(jid, fut)) + status[jid] = State.PENDING else: - print("Fut {} with unknown state".format(fut)) - if fut in self.running_steps: - LOGGER.debug("Fut {} is running?".format(fut)) - LOGGER.debug("Fut {} state: running {}, done {}, cancelled {}".format( - fut, fut.done(), fut.running(), fut.cancelled())) - status[fut] = State.UNKNOWN # this ever reached, and if so how + print("Job {}, with Fut {} with unknown state".format(jid, fut)) + if jid in self.running_steps: + LOGGER.debug("Job {}, with Fut {} is running?".format(jid, fut)) + LOGGER.debug("Job {}, Fut {} state: running {}, done {}, cancelled {}".format( + jid, fut, fut.done(), fut.running(), fut.cancelled())) + status[jid] = State.UNKNOWN # this ever reached, and if so how status_code = JobStatusCode.ERROR return status_code, status @@ -281,29 +300,33 @@ def cancel_jobs(self, joblist): retcode = 0 - for fut in joblist: - if fut in self.running_steps: + for jid in joblist: + if jid in self.running_steps: + fut, pid, step_procs = self.running_steps[jid] if fut.done(): # cleanup - # Really need _submit to return submission record vs retcode/err? - result = fut.get_result() + # Really need _submit to return submission record vs retcode/err? + result = fut.result() + LOGGER.debug("Removing job {} from running steps. result = {}".format(jid, result)) + else: # Interrupt running subprocesses try: - process, procs = self.running_steps[fut] - self._kill(process.pid) #process.kill() # better way to kill it? - fut.result() # wait on future to exit + self._kill(pid) #process.kill() # better way to kill it? + result=fut.result() # wait on future to exit + LOGGER.debug("Removing job {} from running steps. result = {}".format(jid, result)) except KeyError: LOGGER.error("Error, future {}, no longer in running step list".format(fut)) retcode += 1 except: # TODO: catch exceptions from kill(), get_result() LOGGER.exception("Error, unexpected behavior trying to cancel future {}, " - "and subprocess with id {}".format(fut, process.pid)) + "and subprocess with id {}".format(fut, pid)) retcode += 1 - - if not fut.done() and not fut.running(): + + else: # What state occurs when executiongraph knows of jobs that aren't here? + # if not fut.done() and not fut.running(): # This shouldn't be hit since execution graph only submits if resources available - LOGGER.error("Error, encounterd future {} in unexpected state".format(fut)) + LOGGER.error("Error, encounterd job with id {} in unexpected state".format(jid)) retcode += 1 if retcode == 0: @@ -376,6 +399,7 @@ def _submit(self, p, step, path, cwd, job_map=None, env=None): if retcode == 0: LOGGER.info("Execution returned status OK.") + # REPLACE PID WITH UUID? return SubmissionRecord(SubmissionCode.OK, retcode, pid) else: LOGGER.warning("Execution returned an error: %s", str(err)) @@ -419,30 +443,25 @@ def submit(self, step, path, cwd, job_map=None, env=None): LOGGER.warning("Execution returned an error") _record = SubmissionRecord(SubmissionCode.ERROR, -1) - self.avail_procs += step_procs + #self.avail_procs += step_procs return _record # self.avail_procs -= step._procs # except: - # Update running task list. Thread safe list be better? - - self._update_running_steps(add_tasks=[{fut:(p, step_procs)}]) + # Update running task list. + jid = uuid.uuid4() + self._update_running_steps(add_tasks=[{jid: (fut, p.pid, step_procs)}]) # fut.add_done_callback(func_partial(self._task_callback, fut, rm_func=self._update_running_steps)) - fut.add_done_callback(self.wrap_callback(self._update_running_steps)) + fut.add_done_callback(self.wrap_callback(self._update_running_steps, jid)) LOGGER.info("Execution returned status OK.") - return SubmissionRecord(SubmissionCode.OK, 0, fut) + return SubmissionRecord(SubmissionCode.OK, 0, jid) @staticmethod - def wrap_callback(rm_func): + def wrap_callback(rm_func, jid): def task_callback(fut): #fut.result() - rm_func(add_tasks=None, rm_tasks=[fut]) + rm_func(add_tasks=None, rm_tasks=[jid]) return task_callback - - @staticmethod - def _task_callback(fut, rm_func): - fut.get_result() - rm_func(add_tasks=None, rm_tasks=[fut]) From bdc2d0b2b9e94b7115ba484241610fd512c3d649 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Tue, 5 May 2020 23:28:02 -0700 Subject: [PATCH 18/27] Avoid popping ready steps until sure it will run, add early exit of submit loop --- maestrowf/datastructures/core/executiongraph.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index a3b323624..e5a78ba47 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -867,7 +867,7 @@ def execute_ready_steps(self): for i in range(0, _available): # Pop the record and execute using the helper method. - _record = self.values[self.ready_steps.popleft()] + _record = self.values[self.ready_steps[0]] # If we get to this point and we've cancelled, cancel the record. if self.is_canceled: @@ -895,9 +895,13 @@ def execute_ready_steps(self): # NOTE: better place to set this default, and do type conversions? if step_procs <= avail_procs: LOGGER.debug("Launching job %d -- %s", i, _record.name) + self.ready_steps.popleft() # remove key now that it's sure to run self._execute_record(_record, adapter) else: LOGGER.debug("step_procs thought > avail_procs: %d : %d", step_procs, avail_procs) + if avail_procs == 0: + break # exit loop early to avoid needless churn + # check the status of the study upon finishing this round of execution completion_status = self._check_study_completion() From fecc7a745217da5e0f52b87d6b43297c5fc776a8 Mon Sep 17 00:00:00 2001 From: Francesco Di Natale Date: Wed, 13 May 2020 17:54:02 -0700 Subject: [PATCH 19/27] Removal of stray local_procs variables --- maestrowf/datastructures/core/study.py | 1 - 1 file changed, 1 deletion(-) diff --git a/maestrowf/datastructures/core/study.py b/maestrowf/datastructures/core/study.py index 58908b005..e6de13a73 100644 --- a/maestrowf/datastructures/core/study.py +++ b/maestrowf/datastructures/core/study.py @@ -416,7 +416,6 @@ def configure_study(self, submission_attempts=1, restart_limit=1, self._submission_attempts = submission_attempts self._restart_limit = restart_limit self._submission_throttle = throttle - self._local_procs = local_procs self._use_tmp = use_tmp self._hash_ws = hash_ws self._dry_run = dry_run From 1544696f706a7daf3ea249f6bec85adcab04cd72 Mon Sep 17 00:00:00 2001 From: Francesco Di Natale Date: Wed, 13 May 2020 17:58:01 -0700 Subject: [PATCH 20/27] Removal of more stray local_procs --- maestrowf/datastructures/core/executiongraph.py | 2 -- maestrowf/maestro.py | 3 --- 2 files changed, 5 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index e5a78ba47..078ba7da0 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -339,8 +339,6 @@ def __init__(self, submission_attempts=1, submission_throttle=0, self._submission_throttle = submission_throttle self.dry_run = dry_run - self._local_procs = local_procs - # A map that tracks the dependencies of a step. # NOTE: I don't know how performant the Python dict structure is, but # we'll use it for now. I think this may want to be changed to an AVL diff --git a/maestrowf/maestro.py b/maestrowf/maestro.py index 688463dbc..0daf23a79 100644 --- a/maestrowf/maestro.py +++ b/maestrowf/maestro.py @@ -328,9 +328,6 @@ def setup_argparser(): help="Maximum number of inflight jobs allowed to execute " "simultaneously (0 denotes not throttling)." "[Default: %(default)d]") - run.add_argument("-lp", "--local_procs", type=int, default=1, - help="Max number of processors to use simultaneously." - "[Default: %(default)d]") run.add_argument("-s", "--sleeptime", type=int, default=60, help="Amount of time (in seconds) for the manager to " "wait between job status checks. [Default: %(default)d]") From be4ced010e6bad1ebb8bc4304a6a0b8cd2683a7a Mon Sep 17 00:00:00 2001 From: Francesco Di Natale Date: Thu, 14 May 2020 09:52:56 -0700 Subject: [PATCH 21/27] Porting of some missed variables in rebase. --- maestrowf/interfaces/script/localparscriptadapter.py | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index df947280a..38a500db7 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -36,7 +36,6 @@ from concurrent.futures import ThreadPoolExecutor from concurrent.futures._base import PENDING as future_PENDING - from functools import partial as func_partial from threading import Thread, RLock @@ -87,12 +86,8 @@ def __init__(self, **kwargs): } # Maybe need better types here for populating the queues? - self.START = 0 - self.CANCEL = 1 - self.DONE = 2 - - self.total_procs = int(kwargs.pop("proc_count", "1")) - self.avail_procs = int(kwargs.pop("avail_procs", "1")) + self.total_procs = self._batch['proc_count'] + self.avail_procs = self.total_procs self.executor = ThreadPoolExecutor(max_workers=self.total_procs) def __getstate__(self): From 1eae4d520400ad5bfbc3474129d82fca19491ee9 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Fri, 12 Jun 2020 15:53:04 -0700 Subject: [PATCH 22/27] Fix previous broken rebase... --- maestrowf/datastructures/core/executiongraph.py | 1 - maestrowf/datastructures/core/study.py | 1 - maestrowf/maestro.py | 3 --- 3 files changed, 5 deletions(-) diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index e5a78ba47..2b1c6a527 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -339,7 +339,6 @@ def __init__(self, submission_attempts=1, submission_throttle=0, self._submission_throttle = submission_throttle self.dry_run = dry_run - self._local_procs = local_procs # A map that tracks the dependencies of a step. # NOTE: I don't know how performant the Python dict structure is, but diff --git a/maestrowf/datastructures/core/study.py b/maestrowf/datastructures/core/study.py index 58908b005..e6de13a73 100644 --- a/maestrowf/datastructures/core/study.py +++ b/maestrowf/datastructures/core/study.py @@ -416,7 +416,6 @@ def configure_study(self, submission_attempts=1, restart_limit=1, self._submission_attempts = submission_attempts self._restart_limit = restart_limit self._submission_throttle = throttle - self._local_procs = local_procs self._use_tmp = use_tmp self._hash_ws = hash_ws self._dry_run = dry_run diff --git a/maestrowf/maestro.py b/maestrowf/maestro.py index 688463dbc..0daf23a79 100644 --- a/maestrowf/maestro.py +++ b/maestrowf/maestro.py @@ -328,9 +328,6 @@ def setup_argparser(): help="Maximum number of inflight jobs allowed to execute " "simultaneously (0 denotes not throttling)." "[Default: %(default)d]") - run.add_argument("-lp", "--local_procs", type=int, default=1, - help="Max number of processors to use simultaneously." - "[Default: %(default)d]") run.add_argument("-s", "--sleeptime", type=int, default=60, help="Amount of time (in seconds) for the manager to " "wait between job status checks. [Default: %(default)d]") From b2c51affaf15604f7e7ac080c6fa2faf4091e290 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Tue, 1 Feb 2022 19:52:39 -0800 Subject: [PATCH 23/27] Fix incorrect variable --- maestrowf/interfaces/script/localparscriptadapter.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 8c18a45ca..8fa725ea7 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -429,7 +429,7 @@ def submit(self, step, path, cwd, job_map=None, env=None): step_procs = int(step_procs) except ValueError: step_procs = 1 - LOGGER.error("Starting step {} with no 'procs' attribute with 1 processor.".format(_record.step.name)) + LOGGER.error("Starting step {} with no 'procs' attribute with 1 processor.".format(step.name)) self.avail_procs -= step_procs From a4bbd4a53603960c580f9ee9804c98033ac60769 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Mon, 7 Feb 2022 23:59:07 -0800 Subject: [PATCH 24/27] Initial pass hooking up launcher token replacement, fix up broken inheritance --- .../script/localparscriptadapter.py | 248 +++++++++++++++++- 1 file changed, 244 insertions(+), 4 deletions(-) diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 8fa725ea7..68deee8a5 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -31,6 +31,7 @@ import logging import os import psutil +import re import signal import uuid @@ -43,7 +44,6 @@ CancelCode, State from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord from maestrowf.abstracts.interfaces import SchedulerScriptAdapter -from maestrowf.abstracts import Singleton from maestrowf.utils import start_process @@ -62,6 +62,21 @@ class LocalParallelScriptAdapter(SchedulerScriptAdapter): total_procs = 1 avail_procs = 1 + # # The var tag to look for to replace for parallelized commands. + # launcher_var = "$(LAUNCHER)" + # # Allocation regex and compilation + # # Keeping this one here for legacy. + # launcher_regex = re.compile( + # re.escape(launcher_var) + r"\[(?P.*)\]") + + # # We can have multiple requested submission properties. + # # Legacy allocation of nodes and procs. + # legacy_alloc = r"(?P[0-9]+),\s*(?P[0-9]+)" + # # Just allocate based on tasks. + # task_alloc = r"(?P[0-9]+)p" + # # Just allocate based on nodes. + # node_alloc = r"(?P[0-9]+)n" + def __init__(self, **kwargs): """ Initialize an instance of the LocalParallelScriptAdapter. @@ -89,6 +104,16 @@ def __init__(self, **kwargs): self.avail_procs = self.total_procs self.executor = ThreadPoolExecutor(max_workers=self.total_procs) + # Setup initial no-op launcher parameters for default operation + self._cmd_flags = { + "cmd": "", + "ntasks": None, + "nodes": None, + "cores per task": None + } + + self._extension = '.sh' # read this from shell key if present? + def __getstate__(self): """Helper for excluding threadpool from pickling""" state = self.__dict__.copy() @@ -151,9 +176,203 @@ def get_parallelize_command(self, procs, nodes, **kwargs): (default = 1). :returns: A string of the parallelize command configured using nodes and procs. - :NOTE: this is currently a dummy method -> rework to add user specified replacements later + :NOTE: this is currently a dummy method -> rework to add user specified + replacements later + :NOTE: need a mechanism to override these/set from outside adapter """ - return "" + args = [ + self._cmd_flags["cmd"], + ] + + return "".join(args) + + def _substitute_parallel_command(self, step_cmd, **kwargs): + """ + Substitute parallelized segments into a specified command. + + :param step_cmd: Command string to parallelize. + :param nodes: Total number of requested nodes. + :param procs: Total number of requested processors. + :returns: The new command with all allocations substituted. + """ + err_msg = "{} attempting to allocate {} {} for a parallel call with" \ + " a maximum allocation of {}" + + nodes = kwargs.get("nodes") + procs = kwargs.get("procs") + addl_args = dict(kwargs) + addl_args.pop("nodes") + addl_args.pop("procs") + + LOGGER.debug("nodes=%s; procs=%s", nodes, procs) + # See if the command contains a launcher token in it. + alloc_search = list(re.finditer(self.launcher_regex, step_cmd)) + if alloc_search: + # If we find that launcher nomenclature. + total_nodes = 0 # Total nodes we've allocated so far. + total_procs = 0 # Total processors we've allocated so far. + cmd = step_cmd # The step command we'll substitute into. + for match in alloc_search: + LOGGER.debug("Found a match: %s", match.group()) + _nodes = None + _procs = None + # Look for the allocation information in the match. + _alloc = match.group("alloc") + # Search for the legacy format. + _legacy = re.search(self.legacy_alloc, _alloc) + if _legacy: + # nodes, procs legacy notation. + _ = _alloc.split(",") + _nodes = _[0] + _procs = _[1] + LOGGER.debug( + "Legacy setup detected. (nodes=%s, procs=%s)", + _nodes, + _procs + ) + else: + # We're dealing with the new style. + # Make sure we only have at most one proc and node + # allocation specified. + if _alloc.count("p") > 1 or _alloc.count("n") > 1: + msg = "cmd: {}\n Invalid allocations specified ({})." \ + " Number of nodes and/or procs must only be " \ + "specified once." \ + .format(step_cmd, _alloc) + LOGGER.error(msg) + raise ValueError(msg) + + if _alloc.count("p") < 1: + msg = "cmd: {}\n Invalid allocations specified ({})." \ + " Processors/tasks must be specified." \ + .format(step_cmd, _alloc) + LOGGER.error(msg) + raise ValueError(msg) + + _nodes = re.search(self.node_alloc, _alloc) + if _nodes: + _nodes = _nodes.group("nodes") + _procs = re.search(self.task_alloc, _alloc) + if _procs: + _procs = _procs.group("procs") + + LOGGER.debug( + "New setup detected. (nodes=%s, procs=%s)", + _nodes, + _procs + ) + + msg = [] + # Check that the requested nodes are within range. + if _nodes: + _ = int(_nodes) + total_nodes += _ + if _ > nodes: + msg.append( + err_msg.format( + match.group(), _nodes, "nodes", nodes + ) + ) + # Check that the requested processors is within range. + if _procs: + _ = int(_procs) + total_procs += _ + if _ > procs: + msg.append( + err_msg.format( + match.group(), _procs, "procs", procs + ) + ) + # If we have constructed a message, raise an exception. + if msg: + LOGGER.error(msg) + raise ValueError(msg) + + pcmd = self.get_parallelize_command( + _procs, _nodes, **addl_args + ) + cmd = cmd.replace(match.group(), pcmd) + + # Verify that the total nodes/procs used is within maximum. + if total_procs > procs: + msg = "Total processors ({}) requested exceeds the " \ + "maximum requested ({})".format(total_procs, procs) + LOGGER.error(msg) + raise ValueError(msg) + + if total_nodes > nodes: + msg = "Total nodes ({}) requested exceeds the " \ + "maximum requested ({})".format(total_nodes, nodes) + LOGGER.error(msg) + raise ValueError(msg) + + return cmd + else: + # 3. Two smaller cases here. If we see the launcher token WITHOUT + # any parameters, replace it there with full nodes and procs. + # Otherwise, just return the command. A user may simply want to run + # an unparallelized code in a submission. + pcmd = self.get_parallelize_command(procs, nodes, **addl_args) + # Catch the case where the launcher token appears on its own + if self.launcher_var in step_cmd: + LOGGER.debug( + "'%s' found in cmd. Substituting", self.launcher_var) + return step_cmd.replace(self.launcher_var, pcmd) + else: + LOGGER.debug("The command did not specify an MPI command.") + return step_cmd.replace(self.launcher_var, '') + + def get_scheduler_command(self, step): + """ + Generate the full parallelized command for use in a batch script. + + :param step: A StudyStep instance. + :returns: + 1. A Boolean value - True if command is to be scheduled, False + otherwise. + 2. A string representing the parallelized batch command for the + specified step command. + 3. A string representing the parallelized batch command for the + specified step restart command. + """ + # We should never get a study step that doesn't have a run entry; but + # better to be safe. + if not step.run: + msg = "Malformed StudyStep. A StudyStep requires a run entry." + LOGGER.error(msg) + raise ValueError(msg) + + # If the user is requesting nodes, we need to request the nodes and + # set up the command with scheduling. + _nodes = step.run.get("nodes", 0) + _procs = step.run.get("procs", 0) + + to_be_scheduled = False # Local parallel does not submit batch jobs + + if _nodes or _procs: + cmd = self._substitute_parallel_command( + step.run["cmd"], + **step.run + ) + LOGGER.debug("Running parallel command: %s", cmd) + + # Also check for the restart command and parallelize it too. + restart = "" + if step.run["restart"]: + restart = self._substitute_parallel_command( + step.run["restart"], + **step.run + ) + LOGGER.debug("Restart command: %s", cmd) + LOGGER.info("Running parallel workflow step '%s' locally.", step.name) + # Otherwise, just return the command. It doesn't need scheduling. + else: + LOGGER.info("Running workflow step '%s' locally.", step.name) + to_be_scheduled = False + cmd = step.run["cmd"] + restart = step.run["restart"] + + return to_be_scheduled, cmd, restart def _state(self, fut): """ @@ -196,7 +415,8 @@ def _write_script(self, ws_path, step): # THIS IS A HACK FOR NOW: make better use of get_scheduler_command later cmd = step.run["cmd"] restart = step.run["restart"] - to_be_scheduled, _, _ = self.get_scheduler_command(step) + # to_be_scheduled, _, _ = self.get_scheduler_command(step) + to_be_scheduled, cmd, restart = self.get_scheduler_command(step) fname = "{}.sh".format(step.name) script_path = os.path.join(ws_path, fname) @@ -459,3 +679,23 @@ def task_callback(fut): rm_func(add_tasks=None, rm_tasks=[jid]) return task_callback + + @property + def extension(self): + """ + Returns the extension that generated scripts will use. + + :returns: A string of the extension + """ + return self._extension + + + # @property + # def key(self): + # """ + # Return the key name for a ScriptAdapter.. + + # This is used to register the adapter in the ScriptAdapterFactory + # and when writing the workflow specification. + # """ + # return self.key # From 696bf0729d9d170d7f823221d657fb985bcb23d7 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Sun, 27 Feb 2022 21:05:45 -0800 Subject: [PATCH 25/27] Add rich repr to studystep for better debugging --- maestrowf/datastructures/core/study.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/maestrowf/datastructures/core/study.py b/maestrowf/datastructures/core/study.py index 2aea20dfb..d19636b3b 100644 --- a/maestrowf/datastructures/core/study.py +++ b/maestrowf/datastructures/core/study.py @@ -152,6 +152,19 @@ def __ne__(self, other): """ return not self.__eq__(other) + def __rich_repr__(self): + """ + Unique representation of this StudyStep object structure for + pretty printing with the Rich library + + NOTE: does this account for parameter expanded steps? + """ + yield 'name', self.name + yield 'description', self.description + yield 'nickname', self.nickname + yield 'run', self.run + + class Study(DAG, PickleInterface): """ From e863bcaa448d72ffbc3968e7ac18881091d28629 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Sun, 27 Feb 2022 21:07:45 -0800 Subject: [PATCH 26/27] Initial test of script writer --- maestrowf/abstracts/interfaces/__init__.py | 4 +- .../interfaces/schedulerscriptadapter.py | 33 ++++ .../datastructures/core/executiongraph.py | 4 +- maestrowf/interfaces/__init__.py | 54 +++++- .../script/localparscriptadapter.py | 178 ++++-------------- .../interfaces/script/slurmscriptadapter.py | 77 +++++++- .../script/test_slurmscriptadapter.py | 7 +- tests/interfaces/test_script_adapter.py | 35 +++- 8 files changed, 233 insertions(+), 159 deletions(-) diff --git a/maestrowf/abstracts/interfaces/__init__.py b/maestrowf/abstracts/interfaces/__init__.py index e94906a6c..0ac2bebd5 100644 --- a/maestrowf/abstracts/interfaces/__init__.py +++ b/maestrowf/abstracts/interfaces/__init__.py @@ -32,8 +32,8 @@ """ from maestrowf.abstracts.interfaces.schedulerscriptadapter import \ - SchedulerScriptAdapter + SchedulerScriptAdapter, ParallelizeCmd from maestrowf.abstracts.interfaces.scriptadapter import ScriptAdapter -__all__ = ("SchedulerScriptAdapter", "ScriptAdapter") +__all__ = ("SchedulerScriptAdapter", "ScriptAdapter", "ParallelizeCmd") diff --git a/maestrowf/abstracts/interfaces/schedulerscriptadapter.py b/maestrowf/abstracts/interfaces/schedulerscriptadapter.py index 91496b7a2..e6e9ccb96 100644 --- a/maestrowf/abstracts/interfaces/schedulerscriptadapter.py +++ b/maestrowf/abstracts/interfaces/schedulerscriptadapter.py @@ -39,6 +39,21 @@ LOGGER = logging.getLogger(__name__) +@six.add_metaclass(ABCMeta) +class ParallelizeCmd(): + """ + Callable object which generates the concrete scheduler task + launch commands. + """ + def __init__(self, cmd_flags=None, unsupported=None): + self._cmd_flags = {} + self._unsupported = {} + + @abstractmethod + def __call__(self, procs, nodes, **kwargs): + pass + + @six.add_metaclass(ABCMeta) class SchedulerScriptAdapter(ScriptAdapter): """ @@ -82,6 +97,9 @@ def __init__(self, **kwargs): super(SchedulerScriptAdapter, self).__init__(**kwargs) self._batch = {} + # Store launcher -> concrete parallel command mappings + self._cmd_flags = {} + def add_batch_parameter(self, name, value): """ Add a parameter to the ScriptAdapter instance. @@ -92,6 +110,17 @@ def add_batch_parameter(self, name, value): """ self._batch[name] = value + def add_cmd_flags(self, name, value): + """ + Add command flags to the ScriptAdapter instance. Command flags + map the launcher tokens to specific parallel invocations + """ + if name in self._cmd_flags: + LOGGER.info(f"Overwriting '{name}' value ('{self._cmd_flags[name]}" + f"') in cmd_flags with '{value}'") + + self._cmd_flags[name] = value + @abstractmethod def get_header(self, step): """ @@ -103,6 +132,10 @@ def get_header(self, step): """ pass + def register_parallelize_command(self, parallelize_func): + # Note do some validation here -> callable types only + self._parallelize_func = parallelize_func + @abstractmethod def get_parallelize_command(self, procs, nodes, **kwargs): """ diff --git a/maestrowf/datastructures/core/executiongraph.py b/maestrowf/datastructures/core/executiongraph.py index 683767836..e20eb7651 100644 --- a/maestrowf/datastructures/core/executiongraph.py +++ b/maestrowf/datastructures/core/executiongraph.py @@ -536,7 +536,9 @@ def generate_scripts(self): # Set up the adapter. LOGGER.info("Generating scripts...") - adapter = ScriptAdapterFactory.get_adapter(self._adapter["type"]) + # adapter = ScriptAdapterFactory.get_adapter(self._adapter["type"], + # self._adapter.get('parallel_type') + adapter = ScriptAdapterFactory.get_adapter(self._adapter["type"]) adapter = adapter(**self._adapter) self._check_tmp_dir() diff --git a/maestrowf/interfaces/__init__.py b/maestrowf/interfaces/__init__.py index 671e2b1b8..f3e0ddc08 100644 --- a/maestrowf/interfaces/__init__.py +++ b/maestrowf/interfaces/__init__.py @@ -31,8 +31,7 @@ import logging import pkgutil import inspect -from maestrowf.abstracts.interfaces import ScriptAdapter - +from maestrowf.abstracts.interfaces import ScriptAdapter, ParallelizeCmd __all__ = ("ScriptAdapterFactory",) LOGGER = logging.getLogger(__name__) @@ -62,6 +61,42 @@ def iter_adapters(): if isinstance(cls, type) and issubclass(cls, ScriptAdapter) and \ not inspect.isabstract(cls): cs.append(cls) + print("FOUND CLASS '{}' with key '{}'".format(cls.__name__, cls.key)) + if isinstance(cls, type) and issubclass(cls, ScriptAdapter): + print("FOUND CLASS '{}'".format(cls.__name__)) + LOGGER.debug("Found class '{}'".format(cls.__name__)) + + return cs + + +def iter_parallel_cmds(): + """ + Based off of packaging.python.org loop over a namespace and find the + modules. This has been adapted for this particular use case of loading + all classes implementing ParallelizeCmd loaded from all modules in + maestrowf.interfaces.script. + :return: an iterable of the classes existing in the namespace + """ + # get loader for the script adapter package + loader = pkgutil.get_loader('maestrowf.interfaces.script') + # get all of the modules in the package + mods = [(name, ispkg) for finder, name, ispkg in pkgutil.iter_modules( + loader.load_module('maestrowf.interfaces.script').__path__, + loader.load_module('maestrowf.interfaces.script').__name__ + ".")] + + cs = [] + for name, _ in mods: + # get loader for every module + m = pkgutil.get_loader(name).load_module(name) + # get all classes that implement ParallelizeCmd and are not abstract + for n, cls in m.__dict__.items(): + if isinstance(cls, type) and issubclass(cls, ParallelizeCmd) and \ + not inspect.isabstract(cls): + cs.append(cls) + print("FOUND CLASS '{}' with key '{}'".format(cls.__name__, cls.key)) + if isinstance(cls, type) and issubclass(cls, ParallelizeCmd): + print("FOUND CLASS '{}'".format(cls.__name__)) + LOGGER.debug("Found class '{}'".format(cls.__name__)) return cs @@ -71,8 +106,12 @@ class ScriptAdapterFactory(object): adapter.key: adapter for adapter in iter_adapters() } + parallel_cmd_factories = { + parallel_cmd.key: parallel_cmd for parallel_cmd in iter_parallel_cmds() + } + @classmethod - def get_adapter(cls, adapter_id): + def get_adapter(cls, adapter_id, parallel_type=None): if adapter_id.lower() not in cls.factories: msg = "Adapter '{0}' not found. Specify an adapter that exists " \ "or implement a new one mapping to the '{0}'" \ @@ -80,6 +119,15 @@ def get_adapter(cls, adapter_id): LOGGER.error(msg) raise Exception(msg) + # adapter = cls.factories[adapter_id] + + # # Attach any specified parallelize cmd generators + # if parallel_type and adapter_id in cls.parallel_cmd_factories: + # adapter.register_parallelize_command( + # cls.parallel_cmd_factories[adapter_id] + # ) + + print(f"Returning: {cls.factories[adapter_id]}") return cls.factories[adapter_id] @classmethod diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 68deee8a5..50148207b 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -49,6 +49,20 @@ LOGGER = logging.getLogger(__name__) +DEFAULT_LOCAL_CMD_FLAGS = { + "cmd": "", + "ntasks": None, + "nodes": None, + "cores per task": None +} + + +def get_default_local_cmd_flags(): + """ + Returns default cmd flags for local adapters + """ + return DEFAULT_LOCAL_CMD_FLAGS + class LocalParallelScriptAdapter(SchedulerScriptAdapter): """A ScriptAdapter class for interfacing for parallel local execution.""" @@ -91,6 +105,19 @@ def __init__(self, **kwargs): :param **kwargs: A dictionary with default settings for the adapter. """ LOGGER.debug("kwargs\n--------------------------\n%s", kwargs) + print(f"Self type: {type(self)}") + print(f"Super type: {type(LocalParallelScriptAdapter)}") + print(f"self: {self}\nsuper: {LocalParallelScriptAdapter}") + print(f"LocalParallelScriptAdapter subclass of object: {issubclass(LocalParallelScriptAdapter, object)}") + print(f"self subclass of object: {issubclass(self.__class__, object)}") + print(f"self subclass of LocalParallelScriptAdapter: {issubclass(self.__class__, LocalParallelScriptAdapter)}") + print(f"self bases: {type(self).__bases__}") + print(f"LocalParallelScriptAdapter bases: {LocalParallelScriptAdapter.__bases__}") + print(f"self mro: {type(self).__mro__}") + print(f"LocalParallelScriptAdapter mro: {LocalParallelScriptAdapter.__mro__}") + + print(f"self methods: {sorted(self.__dict__.keys())}") + print(f"base methods: {sorted(LocalParallelScriptAdapter.__dict__.keys())}") super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys @@ -105,12 +132,13 @@ def __init__(self, **kwargs): self.executor = ThreadPoolExecutor(max_workers=self.total_procs) # Setup initial no-op launcher parameters for default operation - self._cmd_flags = { - "cmd": "", - "ntasks": None, - "nodes": None, - "cores per task": None - } + # Load cmd flags for launcher token replacement + if "cmd_flags" in kwargs: + cmd_flags = kwargs.pop("cmd_flags") + else: + cmd_flags = get_default_local_cmd_flags() + + self._cmd_flags.update(cmd_flags) self._extension = '.sh' # read this from shell key if present? @@ -178,7 +206,7 @@ def get_parallelize_command(self, procs, nodes, **kwargs): and procs. :NOTE: this is currently a dummy method -> rework to add user specified replacements later - :NOTE: need a mechanism to override these/set from outside adapter + :note: need a mechanism to override these/set from outside adapter """ args = [ self._cmd_flags["cmd"], @@ -186,142 +214,6 @@ def get_parallelize_command(self, procs, nodes, **kwargs): return "".join(args) - def _substitute_parallel_command(self, step_cmd, **kwargs): - """ - Substitute parallelized segments into a specified command. - - :param step_cmd: Command string to parallelize. - :param nodes: Total number of requested nodes. - :param procs: Total number of requested processors. - :returns: The new command with all allocations substituted. - """ - err_msg = "{} attempting to allocate {} {} for a parallel call with" \ - " a maximum allocation of {}" - - nodes = kwargs.get("nodes") - procs = kwargs.get("procs") - addl_args = dict(kwargs) - addl_args.pop("nodes") - addl_args.pop("procs") - - LOGGER.debug("nodes=%s; procs=%s", nodes, procs) - # See if the command contains a launcher token in it. - alloc_search = list(re.finditer(self.launcher_regex, step_cmd)) - if alloc_search: - # If we find that launcher nomenclature. - total_nodes = 0 # Total nodes we've allocated so far. - total_procs = 0 # Total processors we've allocated so far. - cmd = step_cmd # The step command we'll substitute into. - for match in alloc_search: - LOGGER.debug("Found a match: %s", match.group()) - _nodes = None - _procs = None - # Look for the allocation information in the match. - _alloc = match.group("alloc") - # Search for the legacy format. - _legacy = re.search(self.legacy_alloc, _alloc) - if _legacy: - # nodes, procs legacy notation. - _ = _alloc.split(",") - _nodes = _[0] - _procs = _[1] - LOGGER.debug( - "Legacy setup detected. (nodes=%s, procs=%s)", - _nodes, - _procs - ) - else: - # We're dealing with the new style. - # Make sure we only have at most one proc and node - # allocation specified. - if _alloc.count("p") > 1 or _alloc.count("n") > 1: - msg = "cmd: {}\n Invalid allocations specified ({})." \ - " Number of nodes and/or procs must only be " \ - "specified once." \ - .format(step_cmd, _alloc) - LOGGER.error(msg) - raise ValueError(msg) - - if _alloc.count("p") < 1: - msg = "cmd: {}\n Invalid allocations specified ({})." \ - " Processors/tasks must be specified." \ - .format(step_cmd, _alloc) - LOGGER.error(msg) - raise ValueError(msg) - - _nodes = re.search(self.node_alloc, _alloc) - if _nodes: - _nodes = _nodes.group("nodes") - _procs = re.search(self.task_alloc, _alloc) - if _procs: - _procs = _procs.group("procs") - - LOGGER.debug( - "New setup detected. (nodes=%s, procs=%s)", - _nodes, - _procs - ) - - msg = [] - # Check that the requested nodes are within range. - if _nodes: - _ = int(_nodes) - total_nodes += _ - if _ > nodes: - msg.append( - err_msg.format( - match.group(), _nodes, "nodes", nodes - ) - ) - # Check that the requested processors is within range. - if _procs: - _ = int(_procs) - total_procs += _ - if _ > procs: - msg.append( - err_msg.format( - match.group(), _procs, "procs", procs - ) - ) - # If we have constructed a message, raise an exception. - if msg: - LOGGER.error(msg) - raise ValueError(msg) - - pcmd = self.get_parallelize_command( - _procs, _nodes, **addl_args - ) - cmd = cmd.replace(match.group(), pcmd) - - # Verify that the total nodes/procs used is within maximum. - if total_procs > procs: - msg = "Total processors ({}) requested exceeds the " \ - "maximum requested ({})".format(total_procs, procs) - LOGGER.error(msg) - raise ValueError(msg) - - if total_nodes > nodes: - msg = "Total nodes ({}) requested exceeds the " \ - "maximum requested ({})".format(total_nodes, nodes) - LOGGER.error(msg) - raise ValueError(msg) - - return cmd - else: - # 3. Two smaller cases here. If we see the launcher token WITHOUT - # any parameters, replace it there with full nodes and procs. - # Otherwise, just return the command. A user may simply want to run - # an unparallelized code in a submission. - pcmd = self.get_parallelize_command(procs, nodes, **addl_args) - # Catch the case where the launcher token appears on its own - if self.launcher_var in step_cmd: - LOGGER.debug( - "'%s' found in cmd. Substituting", self.launcher_var) - return step_cmd.replace(self.launcher_var, pcmd) - else: - LOGGER.debug("The command did not specify an MPI command.") - return step_cmd.replace(self.launcher_var, '') - def get_scheduler_command(self, step): """ Generate the full parallelized command for use in a batch script. diff --git a/maestrowf/interfaces/script/slurmscriptadapter.py b/maestrowf/interfaces/script/slurmscriptadapter.py index 7c7022e06..55745e545 100644 --- a/maestrowf/interfaces/script/slurmscriptadapter.py +++ b/maestrowf/interfaces/script/slurmscriptadapter.py @@ -33,7 +33,8 @@ import os import re -from maestrowf.abstracts.interfaces import SchedulerScriptAdapter +from maestrowf.abstracts.interfaces import SchedulerScriptAdapter, \ + ParallelizeCmd from maestrowf.abstracts.enums import JobStatusCode, State, SubmissionCode, \ CancelCode from maestrowf.interfaces.script import CancellationRecord, SubmissionRecord @@ -41,6 +42,63 @@ LOGGER = logging.getLogger(__name__) +DEFAULT_SLURM_CMD_FLAGS = { + "cmd": "srun", + "depends": "--dependency", + "ntasks": "-n", + "nodes": "-N", + "cores per task": "-c", +} + + +def get_default_slurm_cmd_flags(): + """ + Returns default cmd flags for slurm adapters + """ + return DEFAULT_SLURM_CMD_FLAGS + + +class SlurmParallelizeCmd(ParallelizeCmd): + key = 'slurm' + + def __call__(self, procs, nodes=None, **kwargs): + """ + Generate the SLURM parallelization segement of the command line. + + :param procs: Number of processors to allocate to the parallel call. + :param nodes: Number of nodes to allocate to the parallel call + (default = 1). + :returns: A string of the parallelize command configured using nodes + and procs. + """ + args = [ + # SLURM srun command + self._cmd_flags["cmd"], + # Processors segment + self._cmd_flags["ntasks"], + str(procs) + ] + + if nodes: + args += [ + self._cmd_flags["nodes"], + str(nodes), + ] + + supported = set(kwargs.keys()) - self._unsupported + for key in supported: + value = kwargs.get(key) + if key not in self._cmd_flags: + LOGGER.warning("'%s' is not supported -- omitted.", key) + continue + if value: + args += [ + self._cmd_flags[key], + "{}".format(str(value)) + ] + + return " ".join(args) + class SlurmScriptAdapter(SchedulerScriptAdapter): """A ScriptAdapter class for interfacing with the SLURM scheduler.""" @@ -99,13 +157,13 @@ def __init__(self, **kwargs): self._exclusive = "#SBATCH --exclusive" self._qos = "#SBATCH --qos={qos}" - self._cmd_flags = { - "cmd": "srun", - "depends": "--dependency", - "ntasks": "-n", - "nodes": "-N", - "cores per task": "-c", - } + # Load cmd flags for launcher token replacement + if "cmd_flags" in kwargs: + cmd_flags = kwargs.pop("cmd_flags") + else: + cmd_flags = get_default_slurm_cmd_flags() + + self._cmd_flags.update(cmd_flags) self._extension = ".slurm.sh" self._unsupported = set(["cmd", "depends", "ntasks", "nodes"]) @@ -174,6 +232,9 @@ def get_parallelize_command(self, procs, nodes=None, **kwargs): :returns: A string of the parallelize command configured using nodes and procs. """ + if self._parallelize_func: + return self._parallelize_func(procs, nodes, **kwargs) + args = [ # SLURM srun command self._cmd_flags["cmd"], diff --git a/tests/interfaces/script/test_slurmscriptadapter.py b/tests/interfaces/script/test_slurmscriptadapter.py index 6a6d3443b..186beaf8d 100644 --- a/tests/interfaces/script/test_slurmscriptadapter.py +++ b/tests/interfaces/script/test_slurmscriptadapter.py @@ -55,7 +55,12 @@ def test_slurm_adapter_in_factory(): """ saf = ScriptAdapterFactory # Make sure SlurmScriptAdapter is in the facotries object - assert(saf.factories[SlurmScriptAdapter.key] == SlurmScriptAdapter) + print(SlurmScriptAdapter) + test_adapter = SlurmScriptAdapter + # assert(saf.factories[SlurmScriptAdapter.key] is test_adapter) + print(id(saf.factories[SlurmScriptAdapter.key])) + print(id(SlurmScriptAdapter)) + assert(saf.factories[SlurmScriptAdapter.key] is SlurmScriptAdapter) # Make sure the SlurmScriptAdapter key is in the valid adapters assert(SlurmScriptAdapter.key in ScriptAdapterFactory.get_valid_adapters()) # Make sure that get_adapter returns the SlurmScriptAdapter when asking diff --git a/tests/interfaces/test_script_adapter.py b/tests/interfaces/test_script_adapter.py index 259a18782..b0919f875 100644 --- a/tests/interfaces/test_script_adapter.py +++ b/tests/interfaces/test_script_adapter.py @@ -34,9 +34,14 @@ as it was converted to dynamically load all ScriptAdapters using a namespace plugin methodology. """ +import os + import pytest from maestrowf.interfaces import ScriptAdapterFactory +from maestrowf.datastructures.core import StudyStep + +from rich.pretty import pprint def test_factory(): @@ -53,7 +58,7 @@ def test_get_valid_adapters(): the results from get_valid_adapters() """ saf = ScriptAdapterFactory - assert(saf.factories.keys() == ScriptAdapterFactory.get_valid_adapters()) + assert(sorted(saf.factories.keys()) == sorted(ScriptAdapterFactory.get_valid_adapters())) def test_adapter_none_found(): @@ -63,3 +68,31 @@ def test_adapter_none_found(): """ with pytest.raises(Exception): ScriptAdapterFactory.get_adapter('empty-adapter') + +def test_adapter_script_generator(): + """ + Tests script generation and launcher token replacement + """ + test_step = StudyStep() + test_step.name = 'test-step' + test_step.description = 'script writer test' + test_step.run = { + 'cmd': ['$(LAUNCHER) echo "Hello, $(NAME)!" > hello_world.txt', + 'sleep 5'], + 'procs': 1 + } + pprint(test_step) + + batch_info = { + 'type': 'local_parallel', + 'proc_count': 4, + 'shell': '/bin/bash' + } + adapter = ScriptAdapterFactory.get_adapter(batch_info['type']) + + adapter2 = adapter(**batch_info) + + print(adapter2) + adapter2.write_script(os.path.abspath('.'), test_step) + + assert(1 == None) From 517d851a4a35f9e3e4f495db7cf37636742b4b31 Mon Sep 17 00:00:00 2001 From: Jeremy White Date: Sun, 27 Feb 2022 21:50:14 -0800 Subject: [PATCH 27/27] Hook up slurm parallelize cmd and enable test of its use in local parallel --- .../abstracts/interfaces/schedulerscriptadapter.py | 2 +- maestrowf/interfaces/script/localparscriptadapter.py | 6 +++++- maestrowf/interfaces/script/slurmscriptadapter.py | 9 +++++++++ tests/interfaces/test_script_adapter.py | 12 ++++++++---- 4 files changed, 23 insertions(+), 6 deletions(-) diff --git a/maestrowf/abstracts/interfaces/schedulerscriptadapter.py b/maestrowf/abstracts/interfaces/schedulerscriptadapter.py index e6e9ccb96..f218b92f1 100644 --- a/maestrowf/abstracts/interfaces/schedulerscriptadapter.py +++ b/maestrowf/abstracts/interfaces/schedulerscriptadapter.py @@ -134,7 +134,7 @@ def get_header(self, step): def register_parallelize_command(self, parallelize_func): # Note do some validation here -> callable types only - self._parallelize_func = parallelize_func + self._parallelize_func = parallelize_func() @abstractmethod def get_parallelize_command(self, procs, nodes, **kwargs): diff --git a/maestrowf/interfaces/script/localparscriptadapter.py b/maestrowf/interfaces/script/localparscriptadapter.py index 50148207b..50026be63 100644 --- a/maestrowf/interfaces/script/localparscriptadapter.py +++ b/maestrowf/interfaces/script/localparscriptadapter.py @@ -118,7 +118,8 @@ def __init__(self, **kwargs): print(f"self methods: {sorted(self.__dict__.keys())}") print(f"base methods: {sorted(LocalParallelScriptAdapter.__dict__.keys())}") - super(LocalParallelScriptAdapter, self).__init__(**kwargs) + super().__init__(**kwargs) + # super(LocalParallelScriptAdapter, self).__init__(**kwargs) # Register keys self.add_batch_parameter("proc_count", int(kwargs.pop("proc_count", "1"))) @@ -208,6 +209,9 @@ def get_parallelize_command(self, procs, nodes, **kwargs): replacements later :note: need a mechanism to override these/set from outside adapter """ + if self._parallelize_func: + return self._parallelize_func(procs, nodes, **kwargs) + args = [ self._cmd_flags["cmd"], ] diff --git a/maestrowf/interfaces/script/slurmscriptadapter.py b/maestrowf/interfaces/script/slurmscriptadapter.py index 55745e545..c9a20a2bd 100644 --- a/maestrowf/interfaces/script/slurmscriptadapter.py +++ b/maestrowf/interfaces/script/slurmscriptadapter.py @@ -61,6 +61,15 @@ def get_default_slurm_cmd_flags(): class SlurmParallelizeCmd(ParallelizeCmd): key = 'slurm' + def __init__(self, cmd_flags=None, unsupported=None): + self._cmd_flags = {} + self._unsupported = set(["cmd", "depends", "ntasks", "nodes"]) + + self._cmd_flags.update(get_default_slurm_cmd_flags()) + + if cmd_flags: + self._cmd_flags.update(cmd_flags) + def __call__(self, procs, nodes=None, **kwargs): """ Generate the SLURM parallelization segement of the command line. diff --git a/tests/interfaces/test_script_adapter.py b/tests/interfaces/test_script_adapter.py index b0919f875..72e1ba7ac 100644 --- a/tests/interfaces/test_script_adapter.py +++ b/tests/interfaces/test_script_adapter.py @@ -39,6 +39,7 @@ import pytest from maestrowf.interfaces import ScriptAdapterFactory +from maestrowf.interfaces.script.slurmscriptadapter import SlurmParallelizeCmd from maestrowf.datastructures.core import StudyStep from rich.pretty import pprint @@ -77,9 +78,11 @@ def test_adapter_script_generator(): test_step.name = 'test-step' test_step.description = 'script writer test' test_step.run = { - 'cmd': ['$(LAUNCHER) echo "Hello, $(NAME)!" > hello_world.txt', - 'sleep 5'], - 'procs': 1 + 'cmd': '\n'.join(['$(LAUNCHER) echo "Hello, $(NAME)!" > hello_world.txt', + 'sleep 5']), + 'procs': 1, + 'nodes': '', + 'restart': '\n'.join(['']) } pprint(test_step) @@ -91,7 +94,8 @@ def test_adapter_script_generator(): adapter = ScriptAdapterFactory.get_adapter(batch_info['type']) adapter2 = adapter(**batch_info) - + adapter2.register_parallelize_command(SlurmParallelizeCmd) + print(adapter2) adapter2.write_script(os.path.abspath('.'), test_step)