From f7c42fa5f8580dc5f5a9a8b7bff204ea321dafaa Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 17:03:46 -0400 Subject: [PATCH 1/6] Phase 6: run a Loop's iterations in parallel New parameters (parallel, iterations at once, cores, memory and time per iteration, snapshot, placement, same-cell writes), shown in the dialog only for a parallel loop. Each iteration runs in an evaluator of its own, resumed into just that iteration; the parent merges them in order, one transaction and checkpoint each. The serial loop is refactored around a set-up per iteration. Co-Authored-By: Claude Opus 5.5 --- loop_step/loop.py | 1117 ++++++++++++++++++++++------------ loop_step/loop_parameters.py | 98 +++ loop_step/tk_loop.py | 27 + tests/test_gui.py | 60 ++ 4 files changed, 923 insertions(+), 379 deletions(-) create mode 100644 tests/test_gui.py diff --git a/loop_step/loop.py b/loop_step/loop.py index 014beb5..1bf6bd8 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -2,11 +2,14 @@ """Non-graphical part of the Loop step in a SEAMM flowchart""" +import json import logging +import os from pathlib import Path import re import shlex import shutil +import sqlite3 import sys import traceback @@ -33,6 +36,29 @@ def _plain(value): printer = printing.getPrinter("loop") +def _yes(value): + """A boolean parameter's value, as text or as a bool.""" + if isinstance(value, str): + return value.strip().lower() in ("yes", "true", "1") + return bool(value) + + +def _in_units(value, units): + """A quantity's magnitude in ``units``; None when not given.""" + if value is None: + return None + if isinstance(value, str): + text = value.strip() + if text == "" or text.startswith("not"): + return None + from seamm_util import Q_ + + value = Q_(text) + if hasattr(value, "to"): + return float(value.to(units).magnitude) + return float(value) + + def _get_decimal_places(value): """Get the number of decimal places in a float.""" if not isinstance(value, float): @@ -236,18 +262,81 @@ def run(self): resume = None if checkpointer is not None: resume = checkpointer.loop_resume(self) + state = None resume_node = None if resume is not None: state, resume_node = resume + + # The evaluator of one iteration of a parallel loop runs just that one + only = state is not None and state.get("only", False) + parallel = not only and self._parallel(P, checkpointer) + if state is not None and state.get("parallel") and not parallel: + raise seamm.CheckpointError( + "This loop was running its iterations in parallel when the job " + "stopped, and cannot resume them one after another." + ) + if state is not None and not state.get("parallel"): self._restore_state(state) # Print out header to the main output printer.important(__(self.description_text(P), indent=self.indent)) - # Set up some unchanging variables - if P["type"] == "For": - # Some local variables need each iteration + context = self._prepare(P, state) + + # Remove any redirection of printing. + if self._file_handler is not None: + job.removeHandler(self._file_handler) + self._file_handler = None + # Find the handler for job.out and set the level up + job_handler = None + out_handler = None + for handler in job.handlers: + if ( + isinstance(handler, logging.FileHandler) + and "job.out" in handler.baseFilename + ): + job_handler = handler + job_level = job_handler.level + job_handler.setLevel(printing.JOB) + elif isinstance(handler, logging.StreamHandler): + out_handler = handler + out_level = out_handler.level + out_handler.setLevel(printing.JOB) + + try: + if parallel: + self._run_parallel(P, context, checkpointer, state) + else: + self._run_serial( + P, context, checkpointer, state, resume_node, only=only + ) + finally: + # Remove any redirection of printing. + if self._file_handler is not None: + self._file_handler.close() + job.removeHandler(self._file_handler) + self._file_handler = None + if job_handler is not None: + job_handler.setLevel(job_level) + if out_handler is not None: + out_handler.setLevel(out_level) + + return self.exit_node() + + # ------------------------------------------------------------------ + # Setting the loop and its iterations up + # ------------------------------------------------------------------ + def _prepare(self, P, state=None): + """What the iterations are: a dict with their number, ``length``. + + On a fresh start (``state`` None) also initializes the loop's variables; + resuming, the items of a loop over rows or systems are the ones in the + checkpoint, frozen when the loop started. + """ + fresh = state is None + context = {} + if P["type"] == "For": # See if loop variables are all integers start = P["start"] if isinstance(start, str): @@ -278,77 +367,50 @@ def run(self): else: fmt = f"0{len(str(end))}d" - if self._loop_value is None: + # The values, as adding the step each iteration gives them (range + # doesn't work for nonintegers) + values = [] + tmp = start + while tmp <= end: + values.append(tmp) + tmp += step + if ndigits > 0: + tmp = round(tmp, ndigits) + context.update( + start=start, step=step, end=end, ndigits=ndigits, fmt=fmt, values=values + ) + context["past"] = tmp + context["length"] = len(values) + if fresh: self.logger.info( "For {} from {} to {} by {}".format( P["variable"], P["start"], P["end"], P["step"] ) ) - self.logger.info("Initializing loop") self._loop_count = 0 self._loop_value = start self.set_variable(P["variable"], self._loop_value) - - # Loop to get length... range doesn't work for nonintegers - count = 0 - tmp = start - while tmp <= end: - count += 1 - tmp += step - if ndigits > 0: - tmp = round(tmp, ndigits) - self._loop_length = count - printer.important( - __( - f"The loop will have {self._loop_length} iterations.\n\n", - indent=self.indent + 4 * " ", - ) - ) - if self.variable_exists("_loop_indices"): - tmp = self.get_variable("_loop_indices") - self.set_variable("_loop_indices", (*tmp, self._loop_value)) - else: - self.set_variable("_loop_indices", (self._loop_value,)) - self.set_variable("_loop_index", self._loop_value) + self._loop_length = len(values) + self._push_loop_indices(self._loop_value) elif P["type"] == "Foreach": - if self._loop_value is None: + if isinstance(P["values"], str): + values = shlex.split(P["values"]) + else: + values = list(P["values"]) + context["values"] = values + context["length"] = len(values) + if fresh: self._loop_value = 0 - if isinstance(P["values"], str): - self._loop_length = len(shlex.split(P["values"])) - else: - self._loop_length = len(P["values"]) - printer.important( - __( - f"The loop will have {self._loop_length} iterations.\n\n", - indent=self.indent + 4 * " ", - ) - ) - if self.variable_exists("_loop_indices"): - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp, - None, - ), - ) - else: - self.set_variable("_loop_indices", (None,)) + self._loop_length = len(values) + self._push_loop_indices(None) elif P["type"] == "For rows in table": - if resume is not None: - self.table = self.get_table(P["table"], create=False) + self.table = self.get_table(P["table"], create=False) + if not fresh: table_rows = state["items"] table_indices = state["indices"] - n_table_indices = len(table_indices) - if n_table_indices > 0: - index_is_int = isinstance(table_indices[0], int) - if index_is_int: - fmt = f"0{len(str(max(table_indices) + 1))}d" - if self._loop_value is None: - self.table = self.get_table(P["table"], create=False) + else: self.table["loop index"] = True - self.logger.info( "Initialize loop over {} rows in table {}".format( self.table.n_rows, P["table"] @@ -356,320 +418,268 @@ def run(self): ) self._loop_value = 0 self._loop_length = self.table.n_rows - if self.variable_exists("_loop_indices"): - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp, - None, - ), - ) - else: - self.set_variable("_loop_indices", (None,)) - where = P["where"] - if where == "Use all rows": - table_rows = [row for row, _ in self.table.rows()] - elif where == "Select rows where column": - op = P["query-op"] - if op not in seamm.table.operators: - raise NotImplementedError(f"Loop query '{op}' not implemented") - try: - selection = ( - P["query-column"], - op, - P["query-value"], - P["query-value2"], - ) - table_rows = [ - row for row, _ in self.table.rows(where=selection) - ] - except ValueError as e: - if "has no column" not in str(e): - raise - column = P["query-column"] - raise ValueError( - f"Looping over table with criterion on column '{column}': " - "that column does not exist." - ) - else: - raise NotImplementedError(f"Loop cannot handle '{where}'") + self._push_loop_indices(None) + table_rows = self._select_rows(P) table_indices = [self.table.label(row) for row in table_rows] - n_table_indices = len(table_indices) - printer.important( - __( - f"The loop will have {n_table_indices} iterations.\n\n", - indent=self.indent + 4 * " ", - ) - ) - - if n_table_indices > 0: - index = table_indices[0] - index_is_int = isinstance(index, int) - if index_is_int: - fmt = f"0{len(str(max(table_indices) + 1))}d" + n_table_indices = len(table_indices) + index_is_int = False + fmt = None + if n_table_indices > 0: + index_is_int = isinstance(table_indices[0], int) + if index_is_int: + fmt = f"0{len(str(max(table_indices) + 1))}d" + context.update( + rows=table_rows, + indices=table_indices, + index_is_int=index_is_int, + fmt=fmt, + length=n_table_indices, + ) elif P["type"] == "For systems in the database": # The configurations to loop over: the standard SEAMM selection. An # empty selection is simply a loop with no iterations. - if resume is not None: + if not fresh: system_db = self.get_variable("_system_db") configurations = [ system_db.get_configuration(cid) for cid in state["items"] ] else: configurations = self.select_configurations(P, errors=False) - - if self._loop_value is None: self._loop_value = 0 self._loop_length = len(configurations) - printer.important( - __( - f"The loop will have {self._loop_length} iterations.\n\n", - indent=self.indent + 4 * " ", - ) + self._push_loop_indices(None) + context["configurations"] = configurations + context["length"] = len(configurations) + else: + raise RuntimeError(f"Don't recognize the loop type {P['type']}") + if fresh: + printer.important( + __( + f"The loop will have {context['length']} iterations.\n\n", + indent=self.indent + 4 * " ", ) - if self.variable_exists("_loop_indices"): - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp, - None, - ), - ) - else: - self.set_variable("_loop_indices", (None,)) + ) + return context - # Remove any redirection of printing. + def _push_loop_indices(self, value): + """Add this loop's index to the loop indices of the enclosing loops.""" + if self.variable_exists("_loop_indices"): + tmp = self.get_variable("_loop_indices") + self.set_variable("_loop_indices", (*tmp, value)) + else: + self.set_variable("_loop_indices", (value,)) + if value is not None: + self.set_variable("_loop_index", value) + + def _pop_loop_indices(self): + """Revert the loop index variables to the next outer loop, if any.""" + tmp = self.get_variable("_loop_indices") + if len(tmp) <= 1: + self.delete_variable("_loop_indices") + self.delete_variable("_loop_index") + else: + self.set_variable("_loop_indices", tmp[0:-1]) + self.set_variable("_loop_index", tmp[-2]) + + def _set_loop_index(self, value): + tmp = self.get_variable("_loop_indices") + self.set_variable("_loop_indices", (*tmp[0:-1], value)) + self.set_variable("_loop_index", value) + + def _select_rows(self, P): + """The rows of the table to loop over.""" + where = P["where"] + if where == "Use all rows": + return [row for row, _ in self.table.rows()] + if where == "Select rows where column": + op = P["query-op"] + if op not in seamm.table.operators: + raise NotImplementedError(f"Loop query '{op}' not implemented") + try: + selection = ( + P["query-column"], + op, + P["query-value"], + P["query-value2"], + ) + return [row for row, _ in self.table.rows(where=selection)] + except ValueError as e: + if "has no column" not in str(e): + raise + column = P["query-column"] + raise ValueError( + f"Looping over table with criterion on column '{column}': " + "that column does not exist." + ) + raise NotImplementedError(f"Loop cannot handle '{where}'") + + def _iteration_number(self, P): + """The number (1, 2, ...) of the current iteration, 0 before the first.""" + if P["type"] == "For": + return self._loop_count + return self._loop_value + + def _setup_iteration(self, P, context, k): + """Set the loop's variables, directory name, etc. for iteration ``k``.""" + loop_type = P["type"] + self._custom_directory_name = None + if loop_type == "For": + self._loop_count = k + self._loop_value = context["values"][k - 1] + self.set_variable(P["variable"], self._loop_value) + # Use the value for the directory names + self._custom_directory_name = f"iter_{self._loop_value:{context['fmt']}}" + self._set_loop_index(self._loop_value) + self.logger.info(" Loop value = {}".format(self._loop_value)) + elif loop_type == "Foreach": + self.logger.debug(f"Foreach {P['variable']} in {P['values']}") + self._loop_value = k + value = context["values"][k - 1] + self.set_variable(P["variable"], value) + self._set_loop_index(self._loop_value) + self.logger.info(" Loop value = {}".format(value)) + elif loop_type == "For rows in table": + self._loop_value = k + index = context["indices"][k - 1] + self._set_loop_index(index) + self.table.current_row = context["rows"][k - 1] + + # Name of directory is the index (+1 since tends to be 0 based) + if context["index_is_int"]: + self._custom_directory_name = f"iter_{index + 1:{context['fmt']}}" + else: + self._custom_directory_name = self.safe_filename(str(index)) + + row = { + k: v + for k, v in self.table.get_row().items() + if k != self.table.index_column + } + self.set_variable("_row", row) + if P["as variables"]: + for key, value in row.items(): + # Make a safe variable name + key = re.sub(r"[-\\ / \+\*()]", "_", key) + self.set_variable(key, value) + self.logger.debug(" _row = {}".format(row)) + elif loop_type == "For systems in the database": + self._loop_value = k + # Set the default system and configuration + configuration = context["configurations"][k - 1] + system_db = configuration.system_db + system = configuration.system + system_db.system = configuration.system + system.configuration = configuration + + if P["directory name"] == "system name": + self._custom_directory_name = self.safe_filename(system.name) + elif P["directory name"] == "configuration name": + self._custom_directory_name = self.safe_filename(configuration.name) + else: + self._custom_directory_name = None + self._set_loop_index(self._loop_value) + self.logger.info(f" system = {system.name}") + self.logger.info(f"configuration = {configuration.name}") + + def _end_loop(self, P, context): + """Tidy the loop's variables away after its last iteration.""" + loop_type = P["type"] + if loop_type == "For": + # As always: the loop variable is left one step past the end + self.set_variable(P["variable"], context["past"]) + self._loop_value = None + self._custom_directory_name = None + self._pop_loop_indices() + self.logger.info( + f"The loop over {P['variable']} from {context['start']} to " + f"{context['end']} by {context['step']} finished successfully" + ) + elif loop_type == "Foreach": + self._loop_value = None + self._loop_length = None + self._pop_loop_indices() + self.logger.info("The loop over value finished successfully") + elif loop_type == "For rows in table": + self._loop_value = None + self.delete_variable("_row") + self._pop_loop_indices() + # and the other info in the table + self.table["loop index"] = False + self.table = None + self.logger.info( + "The loop over table " + + self.parameters["table"].value + + " finished successfully" + ) + elif loop_type == "For systems in the database": + self._loop_value = None + self._loop_length = None + self._custom_directory_name = None + self._pop_loop_indices() + self.logger.info("The loop over value finished successfully") + + def _items(self, P, context): + """The frozen items of the loop, for the checkpoint.""" + if P["type"] == "For rows in table": + return (context["rows"], context["indices"]) + elif P["type"] == "For systems in the database": + return ([c.id for c in context["configurations"]], None) + return (None, None) + + def _open_iteration_output(self, iter_dir, keep): + """Direct most output to the iteration's iteration.out.""" + if self._file_handler is not None: + self._file_handler.close() + job.removeHandler(self._file_handler) + path = iter_dir / "iteration.out" + if not keep: + path.unlink(missing_ok=True) + self._file_handler = logging.FileHandler(path) + self._file_handler.setLevel(printing.NORMAL) + formatter = logging.Formatter(fmt="{message:s}", style="{") + self._file_handler.setFormatter(formatter) + job.addHandler(self._file_handler) + + def _close_iteration_output(self): if self._file_handler is not None: + self._file_handler.close() job.removeHandler(self._file_handler) self._file_handler = None - # Find the handler for job.out and set the level up - job_handler = None - out_handler = None - for handler in job.handlers: - if ( - isinstance(handler, logging.FileHandler) - and "job.out" in handler.baseFilename - ): - job_handler = handler - job_level = job_handler.level - job_handler.setLevel(printing.JOB) - elif isinstance(handler, logging.StreamHandler): - out_handler = handler - out_level = out_handler.level - out_handler.setLevel(printing.JOB) + # ------------------------------------------------------------------ + # One iteration after another + # ------------------------------------------------------------------ + def _run_serial(self, P, context, checkpointer, state, resume_node, only=False): + """Run the iterations in this evaluator, one after another. + + ``only``: this evaluator runs one iteration of a parallel loop, the one + it resumed into, and stops when it ends (seamm.IterationDone). + """ + length = context["length"] + resuming = state is not None + if only and resume_node is None: + # The iteration had finished, but the evaluator had not. + raise seamm.IterationDone() # Cycle through the iterations. Resuming in the middle of an iteration, # set it up again without advancing, then start its body at the step # that had not finished. next_node = self - advance = resume is None or resume_node is None + advance = not resuming or resume_node is None + ran = False + iter_dir = None while next_node is not None: if next_node is self: + if only and ran: + raise seamm.IterationDone() next_node = self.loop_node() - - if P["type"] == "For": - if advance: - self._loop_count += 1 - if self._loop_count > 1: - self._loop_value += step - if ndigits > 0: - self._loop_value = round(self._loop_value, ndigits) - - self.set_variable(P["variable"], self._loop_value) - - # Use the value for the directory names - self._custom_directory_name = f"iter_{self._loop_value:{fmt}}" - - # Set up the index variables - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp[0:-1], - self._loop_value, - ), - ) - self.set_variable("_loop_index", self._loop_value) - - # See if we are at the end of loop - if self._loop_count > self._loop_length: - self._loop_value = None - self._custom_directory_name = None - - # Revert the loop index variables to the next outer loop - # if there is one, or remove them. - tmp = self.get_variable("_loop_indices") - - if len(tmp) <= 1: - self.delete_variable("_loop_indices") - self.delete_variable("_loop_index") - else: - self.set_variable("_loop_indices", tmp[0:-1]) - self.set_variable("_loop_index", tmp[-2]) - - self.logger.info( - f"The loop over {P['variable']} from {start} to " - f"{end} by {step} finished successfully" - ) - break - - self.logger.info(" Loop value = {}".format(self._loop_value)) - elif P["type"] == "Foreach": - self.logger.debug(f"Foreach {P['variable']} in {P['values']}") - - if advance: - self._loop_value += 1 - - if self._loop_value > self._loop_length: - self._loop_value = None - self._loop_length = None - - # Revert the loop index variables to the next outer loop - # if there is one, or remove them. - tmp = self.get_variable("_loop_indices") - if len(tmp) <= 1: - self.delete_variable("_loop_indices") - self.delete_variable("_loop_index") - else: - self.set_variable("_loop_indices", tmp[0:-1]) - self.set_variable("_loop_index", tmp[-2]) - self.logger.info("The loop over value finished successfully") - - # return the next node after the loop - break - - if isinstance(P["values"], str): - value = shlex.split(P["values"])[self._loop_value - 1] - else: - value = P["values"][self._loop_value - 1] - self.set_variable(P["variable"], value) - - # Set up the index variables - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp[0:-1], - self._loop_value, - ), - ) - self.set_variable("_loop_index", self._loop_value) - self.logger.info(" Loop value = {}".format(value)) - elif P["type"] == "For rows in table": - if advance: - self._loop_value += 1 - if self._loop_value > n_table_indices: - self._loop_value = None - - self.delete_variable("_row") - # Revert the loop index variables to the next outer loop - # if there is one, or remove them. - tmp = self.get_variable("_loop_indices") - if len(tmp) <= 1: - self.delete_variable("_loop_indices") - self.delete_variable("_loop_index") - else: - self.set_variable("_loop_indices", tmp[0:-1]) - self.set_variable("_loop_index", tmp[-2]) - - # and the other info in the table - self.table["loop index"] = False - - self.table = None - - self.logger.info( - "The loop over table " - + self.parameters["table"].value - + " finished successfully" - ) - - # return the next node after the loop - break - - # Set up the index variables - self.logger.debug(" _loop_value = {}".format(self._loop_value)) - index = table_indices[self._loop_value - 1] - tmp = self.get_variable("_loop_indices") - self.logger.debug(" _loop_indices = {}".format(tmp)) - self.set_variable("_loop_indices", (*tmp[0:-1], index)) - self.logger.debug( - " --> {}".format(self.get_variable("_loop_indices")) - ) - self.set_variable("_loop_index", index) - self.table.current_row = table_rows[self._loop_value - 1] - - # Name of directory is the index (+1 since tends to be 0 based) - if index_is_int: - self._custom_directory_name = f"iter_{index + 1:{fmt}}" - else: - self._custom_directory_name = self.safe_filename(str(index)) - - row = { - k: v - for k, v in self.table.get_row().items() - if k != self.table.index_column - } - self.set_variable("_row", row) - if P["as variables"]: - for key, value in row.items(): - # Make a safe variable name - key = re.sub(r"[-\\ / \+\*()]", "_", key) - self.set_variable(key, value) - self.logger.debug(" _row = {}".format(row)) - elif P["type"] == "For systems in the database": - if advance: - self._loop_value += 1 - - if self._loop_value > self._loop_length: - self._loop_value = None - self._loop_length = None - self._custom_directory_name = None - - # Revert the loop index variables to the next outer loop - # if there is one, or remove them. - tmp = self.get_variable("_loop_indices") - if len(tmp) <= 1: - self.delete_variable("_loop_indices") - self.delete_variable("_loop_index") - else: - self.set_variable("_loop_indices", tmp[0:-1]) - self.set_variable("_loop_index", tmp[-2]) - self.logger.info("The loop over value finished successfully") - - # return the next node after the loop - break - - # Set the default system and configuration - configuration = configurations[self._loop_value - 1] - system_db = configuration.system_db - system = configuration.system - system_db.system = configuration.system - system.configuration = configuration - - if P["directory name"] == "system name": - self._custom_directory_name = self.safe_filename(system.name) - elif P["directory name"] == "configuration name": - self._custom_directory_name = self.safe_filename( - configuration.name - ) - else: - self._custom_directory_name = None - - # Set up the index variables - tmp = self.get_variable("_loop_indices") - self.set_variable( - "_loop_indices", - ( - *tmp[0:-1], - self._loop_value, - ), - ) - self.set_variable("_loop_index", self._loop_value) - self.logger.info(f" system = {system.name}") - self.logger.info(f"configuration = {configuration.name}") + k = self._iteration_number(P) + if advance: + k += 1 + if k > length: + self._end_loop(P, context) + break + self._setup_iteration(P, context, k) # Resuming this iteration: its directory is the one it had, not # a new unique name next to it. @@ -688,34 +698,19 @@ def run(self): if checkpointer is not None: if not advance: next_node = self._find_body_node(resume_node) - if P["type"] == "For rows in table": - items = (table_rows, table_indices) - elif P["type"] == "For systems in the database": - items = ([c.id for c in configurations], None) - else: - items = (None, None) - checkpointer.enter_iteration( - self, self._checkpoint_state(P, *items), next_node + iteration_state = self._checkpoint_state( + P, *self._items(P, context) ) + if only: + iteration_state.update(only=True, done=False) + checkpointer.enter_iteration(self, iteration_state, next_node) resuming_iteration = not advance advance = True + ran = True - # Direct most output to iteration.out - # A handler for the file iter_dir = self.working_path iter_dir.mkdir(parents=True, exist_ok=True) - - if self._file_handler is not None: - self._file_handler.close() - job.removeHandler(self._file_handler) - path = iter_dir / "iteration.out" - if not resuming_iteration: - path.unlink(missing_ok=True) - self._file_handler = logging.FileHandler(path) - self._file_handler.setLevel(printing.NORMAL) - formatter = logging.Formatter(fmt="{message:s}", style="{") - self._file_handler.setFormatter(formatter) - job.addHandler(self._file_handler) + self._open_iteration_output(iter_dir, keep=resuming_iteration) # Run through the steps in the loop body try: @@ -727,10 +722,15 @@ def run(self): traceback.print_exc(file=sys.stderr) traceback.print_exc(file=sys.stdout) except BreakLoop: + if only: + raise seamm.IterationDone(broke=True) break except ContinueLoop: next_node = self except SkipIteration: + if only: + # The parent removes the iteration's directory + raise seamm.IterationDone(skipped=True) next_node = self shutil.rmtree(iter_dir, ignore_errors=True) except Exception as e: @@ -738,6 +738,9 @@ def run(self): printer.job(f"Caught exception in loop iteration {tmp}: {str(e)}") with open(iter_dir / "stderr.out", "a") as fd: traceback.print_exc(file=fd) + if only: + # The parent decides what a failed iteration means + raise if "continue" in P["errors"]: next_node = self elif "exit" in P["errors"]: @@ -761,17 +764,373 @@ def run(self): if checkpointer is not None: checkpointer.leave_loop(self) - # Remove any redirection of printing. - if self._file_handler is not None: - self._file_handler.close() - job.removeHandler(self._file_handler) - self._file_handler = None - if job_handler is not None: - job_handler.setLevel(job_level) - if out_handler is not None: - out_handler.setLevel(out_level) + # ------------------------------------------------------------------ + # Iterations in parallel + # ------------------------------------------------------------------ + def _parallel(self, P, checkpointer): + """Whether to run the iterations in parallel.""" + if not _yes(P["parallel"]): + return False + if checkpointer is None: + printer.important( + __( + "The iterations run one after another: running them in " + "parallel needs the job's database in a file in the job, with " + "checkpoints.", + indent=self.indent + 4 * " ", + ) + ) + return False + return True + + def _job_database_path(self, system_db): + """The file of the job's database.""" + name = system_db.filename + if name.startswith("file:"): + name = name[5:] + name = name.split("?")[0] + path = Path(name) + if not path.is_absolute(): + path = Path(self.flowchart.job_directory) / path + return path + + def _run_parallel(self, P, context, checkpointer, state): + """Run each iteration in an evaluator of its own, then merge them in order. + + See the phase 6 notes in seamm_exec's developer guide. The Loop's + checkpoint frame holds the frozen items, the directories given out, the + next iteration to merge, the failed ones, and what the merge needs from + one iteration to the next; each iteration is merged in one transaction, + committed with that frame. + """ + from molsystem.snapshot import baseline, snapshot + from seamm_exec import LocalPool, TaskSet + from seamm_exec import iteration as iterations + + length = context["length"] + system_db = self.get_variable("_system_db") + loop_directory = Path(self.directory) + job_directory = Path(self.flowchart.job_directory) + first = self.loop_node() + if self._loop_length is None: + self._loop_length = length + later_wins = "later" in P["same-cell writes"] + + if state is None: + items, indices = self._items(P, context) + current = None + if system_db.system is not None: + configuration = system_db.system.configuration + if configuration is not None: + current = configuration.id + state = { + "type": P["type"], + "parallel": True, + "count": 0, + "length": length, + "next": 1, + "directories": {}, + "failed": [], + "merge": {}, + "exports": {}, + "current": current, + "directory": None, + "loop_count": None, + "loop_value": None, + } + if items is not None: + state["items"] = [_plain(x) for x in items] + if indices is not None: + state["indices"] = [_plain(x) for x in indices] + else: + state = json.loads(json.dumps(state)) + printer.important( + __( + f"Resuming the parallel loop: {state['next'] - 1} of {length} " + "iterations were done and merged.", + indent=self.indent + 4 * " ", + ) + ) + merge_state = iterations.decode_merge_state(state.get("merge")) + + # Commit the loop's set-up with its frame, then keep the database as the + # loop starts: every iteration's snapshot is taken from it. + checkpointer.parallel_loop(self, state) + entry = loop_directory / "loop_entry.db" + if state["next"] <= length and not entry.exists(): + loop_directory.mkdir(parents=True, exist_ok=True) + tmp = entry.with_name(entry.name + ".tmp") + snapshot(self._job_database_path(system_db), tmp, whole=True) + # Only ever read: no write-ahead log files beside it + db = sqlite3.connect(str(tmp)) + try: + db.execute("PRAGMA journal_mode=DELETE") + finally: + db.close() + os.replace(tmp, entry) + + # A merge stopped after its database part: finish its files + pending = state.pop("files", None) + if pending is not None: + self._merge_iteration_files(pending, loop_directory, job_directory) + + # The resources of each iteration + cores = max(1, int(P["cores per iteration"])) + memory = _in_units(P["memory per iteration"], "byte") + walltime = _in_units(P["time per iteration"], "s") + at_once = P["iterations at once"] + if isinstance(at_once, str): + at_once = None if "as many" in at_once else int(at_once) + placement = "inline" if P["placement"] == "inline" else "separate" + whole = P["snapshot"] == "whole database" + root = None + try: + root = self.global_options.get("root") + except Exception: + pass + pool = LocalPool(self.flowchart.executor, root=root, max_concurrent=at_once) + task_set = TaskSet(self, local=pool, root=root) + + # A loop that had stopped (a break, or an error) dispatches nothing more + stopped = None + if state.get("stopped") is not None: + stopped = tuple(state["stopped"]) + last = length if stopped is None else 0 + + # Set each iteration up, write its snapshot and checkpoint, add its task + keys = {} + for k in range(state["next"], last + 1): + self._setup_iteration(P, context, k) + name = state["directories"].get(str(k)) + if name is None: + # On record before the directory exists, as for a serial loop + name = self.working_path.name + state["directories"][str(k)] = name + checkpointer.parallel_loop(self, state) + self._custom_directory_name = name + self.flowchart.reset_visited() + self.set_subids((*self._id, name)) + + evaluator = loop_directory / name / iterations.EVALUATOR_DIRECTORY + if not (evaluator / "seamm.db").exists(): + shutil.rmtree(evaluator, ignore_errors=True) + evaluator.mkdir(parents=True) + configurations = [state["current"]] + if P["type"] == "For systems in the database": + configurations.append(context["configurations"][k - 1].id) + tmp = evaluator / "seamm.db.tmp" + snapshot( + entry, + tmp, + configurations=[c for c in configurations if c is not None], + whole=whole, + ) + checkpointer.write_child( + tmp, + self, + self._checkpoint_state(P, *self._items(P, context)), + first, + ) + baseline(tmp, evaluator / "baseline.db") + os.replace(tmp, evaluator / "seamm.db") + task = iterations.iteration_task( + f"iteration_{k}", + evaluator, + job_directory, + command_line=checkpointer.command_line, + cores=cores, + memory=memory, + walltime=walltime, + placement=placement, + root_directory=self.flowchart.root_directory, + ) + task_set.add(task) + keys[task.key] = k + + n = len(keys) + if n > 0: + printer.important( + __( + f"Running {n} iterations in parallel, each with {cores} cores " + f"and {P['memory per iteration']} of memory.", + indent=self.indent + 4 * " ", + ) + ) - return self.exit_node() + # Merge the iterations in order as they finish + results = {} + runs = task_set.run() if n > 0 else iter(()) + try: + for result in runs: + results[keys[result.key]] = result + while state["next"] in results: + k = state["next"] + stopped = self._merge_iteration( + P, + context, + state, + merge_state, + k, + results.pop(k), + checkpointer, + system_db, + loop_directory, + job_directory, + later_wins, + ) + if stopped is not None: + break + if stopped is not None: + break + finally: + # Cancels the iterations still running + if n > 0: + runs.close() + + if stopped is not None: + k, why = stopped + unmerged = sorted(results) + [ + keys[key] + for key in task_set.tasks + if keys[key] > k and keys[key] not in results + ] + if len(unmerged) > 0: + names = ", ".join( + state["directories"][str(i)] for i in sorted(unmerged) + ) + printer.important( + __( + f"The iterations after {state['directories'][str(k)]} are " + f"not merged; their directories are kept: {names}.", + indent=self.indent + 4 * " ", + ) + ) + else: + # Tables the iterations exported, written again from the merged ones + for name, filename in state.get("exports", {}).items(): + try: + self.get_table(name, create=False).export(filename) + except Exception as e: + printer.important(f"Could not export the table '{name}': {e}") + for suffix in ("", "-wal", "-shm"): + Path(str(entry) + suffix).unlink(missing_ok=True) + + n_failed = len(state["failed"]) + text = f"Merged {state['next'] - 1 - n_failed} of {length} iterations" + if n_failed > 0: + failed = ", ".join(state["directories"][str(i)] for i in state["failed"]) + text += f"; {n_failed} failed and were not merged: {failed}" + printer.important(__(text + ".", indent=self.indent + 4 * " ")) + + if stopped is None: + self._end_loop(P, context) + else: + k, why = stopped + # The loop's variables as the iteration that ended it left them + self._setup_iteration(P, context, k) + if why == "raise": + raise RuntimeError( + f"Iteration {state['directories'][str(k)]} of the loop failed." + ) + checkpointer.leave_loop(self) + + def _merge_iteration( + self, + P, + context, + state, + merge_state, + k, + result, + checkpointer, + system_db, + loop_directory, + job_directory, + later_wins, + ): + """Merge iteration ``k``, or record that it failed. + + Returns + ------- + (int, str) or None + ``(k, why)`` if the loop stops here ("break", "exit" or "raise"). + """ + from seamm_exec import iteration as iterations + + name = state["directories"][str(k)] + iter_dir = loop_directory / name + evaluator = iter_dir / iterations.EVALUATOR_DIRECTORY + outcome = iterations.iteration_outcome(evaluator) if result.ok else None + if outcome is None: + reason = result.reason or f"return code {result.returncode}" + printer.job( + f"Caught an error in loop iteration {name} ({reason}); see " + f"{evaluator / 'job.out'}." + ) + if "stop" in P["errors"]: + # The job fails here; a resume runs the iteration again + return (k, "raise") + state["failed"].append(k) + state["next"] = k + 1 + state["count"] = k + if "continue" in P["errors"]: + checkpointer.parallel_loop(self, state) + return None + state["stopped"] = [k, "exit"] + checkpointer.parallel_loop(self, state) + return (k, "exit") + + merged = iterations.merge_database( + system_db, evaluator, merge_state, k, later_wins=later_wins + ) + if P["type"] == "For rows in table": + # As a serial loop leaves it: the iteration's row, unless its body + # moved on (below). The merge cannot see a move to the row that was + # current when the loop started. + self.table.current_row = context["rows"][k - 1] + for table, row in merged["current_rows"].items(): + system_db.user_tables[table].current_row = row + for table, filename in merged["exports"].items(): + state["exports"][table] = str( + iterations.export_path(filename, evaluator, job_directory) + ) + system_id = outcome.get("system_id") + if system_id is not None: + system_id = merged["maps"].get("system", {}).get(system_id, system_id) + if system_id in system_db.system_ids: + system_db.system = system_id + plan = iterations.plan_files(evaluator, job_directory, outcome.get("run_id")) + state["files"] = {"iteration": k, "directory": name, "plan": plan} + state["next"] = k + 1 + state["count"] = k + state["merge"] = iterations.encode_merge_state(merge_state) + stop = None + if outcome.get("break"): + stop = (k, "break") + state["stopped"] = [k, "break"] + checkpointer.parallel_loop(self, state) + self._merge_iteration_files(state.pop("files"), loop_directory, job_directory) + if outcome.get("skip"): + shutil.rmtree(iter_dir, ignore_errors=True) + printer.job(f" Merged loop iteration {name}.") + return stop + + def _merge_iteration_files(self, pending, loop_directory, job_directory): + """The job-level files and citations of a merged iteration. + + ``pending`` is the record in the Loop's frame, committed with the + iteration's database merge; doing it twice gives the same files. + """ + from seamm_exec import iteration as iterations + + evaluator = ( + loop_directory / pending["directory"] / iterations.EVALUATOR_DIRECTORY + ) + if not evaluator.exists(): + # Merged and removed (a skipped iteration) before the job stopped + return + iterations.merge_files(evaluator, job_directory, pending["plan"]) + iterations.merge_citations(evaluator, self.references) def _checkpoint_state(self, P, items=None, indices=None): """The loop's state at the start of an iteration, for the checkpoint.""" diff --git a/loop_step/loop_parameters.py b/loop_step/loop_parameters.py index 5363e5d..636c105 100644 --- a/loop_step/loop_parameters.py +++ b/loop_step/loop_parameters.py @@ -204,6 +204,104 @@ class LoopParameters(seamm.Parameters): "description": "On errors", "help_text": ("How to handle errors"), }, + # Parallel iterations (phase 6 of the parallel-execution campaign) + "parallel": { + "default": "no", + "kind": "boolean", + "default_units": "", + "enumeration": ("no", "yes"), + "format_string": "", + "description": "Run iterations in parallel:", + "help_text": ( + "Run the iterations at the same time, each in its own evaluator. " + "They must be independent: each sees the job as it was when the " + "loop started, and variables set in the body are not visible " + "after the loop. Tables, properties, structures and files come " + "back, merged in iteration order." + ), + }, + "iterations at once": { + "default": "as many as fit", + "kind": "integer", + "default_units": "", + "enumeration": ("as many as fit",), + "format_string": "", + "description": "Iterations at once:", + "help_text": ( + "How many iterations run at the same time on this machine. By " + "default as many as the cores and memory allow, given the cores " + "and memory per iteration." + ), + }, + "cores per iteration": { + "default": 1, + "kind": "integer", + "default_units": "", + "enumeration": tuple(), + "format_string": "", + "description": "Cores per iteration:", + "help_text": "The cores each iteration uses for its calculations.", + }, + "memory per iteration": { + "default": 2.0, + "kind": "float", + "default_units": "GB", + "enumeration": tuple(), + "format_string": ".1f", + "description": "Memory per iteration:", + "help_text": "The memory each iteration needs at most.", + }, + "time per iteration": { + "default": "not given", + "kind": "float", + "default_units": "h", + "enumeration": ("not given",), + "format_string": ".2f", + "description": "Time per iteration:", + "help_text": ( + "An estimate of the time each iteration takes, used to bundle " + "iterations into queue jobs." + ), + }, + "snapshot": { + "default": "selected structures", + "kind": "enumeration", + "default_units": "", + "enumeration": ("selected structures", "whole database"), + "format_string": "s", + "description": "Each iteration gets:", + "help_text": ( + "What of the job's database each iteration starts with: the " + "structure it works on (and the current one) with all the " + "tables, or a copy of the whole database." + ), + }, + "placement": { + "default": "inline", + "kind": "enumeration", + "default_units": "", + "enumeration": ("inline", "separate tasks"), + "format_string": "s", + "description": "Calculations run:", + "help_text": ( + "Where each iteration's calculations run: inline, in the " + "iteration's own share of the machine, or as separate tasks on " + "the job's target." + ), + }, + "same-cell writes": { + "default": "are an error", + "kind": "enumeration", + "default_units": "", + "enumeration": ("are an error", "the later iteration wins"), + "format_string": "s", + "description": "Two iterations writing one table cell:", + "help_text": ( + "What to do when two iterations write the same cell of a table " + "row that existed before the loop: stop with an error, or keep " + "the value from the later iteration, with a warning." + ), + }, } def __init__(self, defaults={}, data=None): diff --git a/loop_step/tk_loop.py b/loop_step/tk_loop.py index ca135ff..4800cca 100644 --- a/loop_step/tk_loop.py +++ b/loop_step/tk_loop.py @@ -49,6 +49,18 @@ } +# The settings of a parallel loop, shown when its iterations run in parallel +PARALLEL_PARAMETERS = ( + "iterations at once", + "cores per iteration", + "memory per iteration", + "time per iteration", + "snapshot", + "placement", + "same-cell writes", +) + + class TkLoop(seamm.TkNode): """The node_class is the class of the 'real' node that this class is the Tk graphics partner for @@ -147,6 +159,12 @@ def create_dialog(self): self["errors"].combobox.config(state="readonly") + # Parallel iterations: their settings only when asked for + self["parallel"].combobox.bind("<>", self.reset_dialog) + self["parallel"].combobox.config(state="readonly") + for widget in ("snapshot", "placement", "same-cell writes"): + self[widget].combobox.config(state="readonly") + def criteria_callback(self, widget, criterion, event, what): """Handle changes in the search criteria widget. @@ -237,6 +255,15 @@ def reset_dialog(self, widget=None): raise RuntimeError("Don't recognize the loop_type {}".format(loop_type)) self["errors"].grid(row=row, column=0, columnspan=4, sticky=tk.W) row += 1 + self["parallel"].grid(row=row, column=0, columnspan=4, sticky=tk.W) + row += 1 + if self["parallel"].get() == "yes": + widgets = [] + for key in PARALLEL_PARAMETERS: + self[key].grid(row=row, column=1, columnspan=4, sticky=tk.W) + widgets.append(self[key]) + row += 1 + sw.align_labels(widgets, sticky=tk.E) frame.columnconfigure(0, minsize=40) def right_click(self, event): diff --git a/tests/test_gui.py b/tests/test_gui.py new file mode 100644 index 0000000..6e76e27 --- /dev/null +++ b/tests/test_gui.py @@ -0,0 +1,60 @@ +# -*- coding: utf-8 -*- + +"""Smoke test of the Tk dialog: create it and re-lay it out for every choice +that drives the layout. Skipped when no display is available.""" + +import pytest + +from loop_step.tk_loop import PARALLEL_PARAMETERS + + +@pytest.fixture() +def tk_node(): + import tkinter as tk + + try: + root = tk.Tk() + except tk.TclError: + pytest.skip("no display available for Tk") + root.withdraw() + import Pmw + import seamm + + Pmw.initialise(root) + flowchart = seamm.Flowchart(namespace="org.molssi.seamm", directory=".") + tk_flowchart = seamm.TkFlowchart( + master=root, flowchart=flowchart, namespace="org.molssi.seamm.tk" + ) + node = flowchart.create_node("Loop") + flowchart.add_node(node) + plugin = tk_flowchart.plugin_manager.get("Loop") + tk_node = plugin.create_tk_node( + tk_flowchart=tk_flowchart, node=node, canvas=tk_flowchart.canvas, x=100, y=100 + ) + yield tk_node + root.destroy() + + +def test_dialog_layouts(tk_node): + tk_node.create_dialog() + tk_node.reset_dialog() + for loop_type in ( + "Foreach", + "For rows in table", + "For systems in the database", + "For", + ): + tk_node["type"].set(loop_type) + for parallel in ("yes", "no"): + tk_node["parallel"].set(parallel) + tk_node.reset_dialog() + shown = [tk_node[k].grid_info() != {} for k in PARALLEL_PARAMETERS] + assert tk_node["parallel"].grid_info() != {} + if parallel == "yes": + assert all(shown), loop_type + else: + assert not any(shown), loop_type + tk_node["type"].set("For rows in table") + for where in ("Use all rows", "Select rows where column"): + tk_node["where"].set(where) + tk_node.reset_dialog() From 92d9ac867c03d28f6fa88c0f9decc180a73aecc7 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 17:44:42 -0400 Subject: [PATCH 2/6] Phase 6: retry iterations killed early; trust the iteration's checkpoint An iteration whose evaluator stopped before it ended (its checkpoint still 'running') runs again, resuming, up to twice; one that finished its iteration is merged even if it was killed as it exited. Co-Authored-By: Claude Opus 5.5 --- docs/user_guide/index.rst | 122 ++++++++++++++++++++++++++++++++++++-- loop_step/loop.py | 102 +++++++++++++++++++++++-------- 2 files changed, 192 insertions(+), 32 deletions(-) diff --git a/docs/user_guide/index.rst b/docs/user_guide/index.rst index ef780d3..0dee7bc 100644 --- a/docs/user_guide/index.rst +++ b/docs/user_guide/index.rst @@ -3,14 +3,124 @@ ********** User Guide ********** -The XXXX plug-in ... +The Loop step runs the steps in its body repeatedly: once for each value of a +variable, each item in a list, each row of a table, or each structure in the +database. -.. - The following sections cover accessing and controlling this functionality. +Types of loops +============== +For + A variable runs from a start to an end value by a step, e.g. ``i`` from 1 to + 10. Each iteration runs in a directory named after the value, + ``iter_01``, ``iter_02``, ... +Foreach + A variable takes each of a list of values in turn, e.g. SMILES strings or + Hamiltonians. +For rows in table + The body runs once for each row of a table, or each row matching a criterion. + The row is the table's current row during the iteration, so steps can read + and write its cells; with "Values as variables" each cell is also a variable. + The rows are chosen when the loop starts: rows the body appends are not + looped over. +For systems in the database + The body runs once for each structure chosen by the standard structure + selection (by default every system, with its last configuration), which is + the current system and configuration during the iteration. Directories may + be named after the systems or configurations. - .. toctree:: - :maxdepth: 2 - :titlesonly: +Errors +====== +"On errors" says what happens when a step in the body fails: continue with the +next iteration (the default), leave the loop and carry on with the rest of the +flowchart, or stop the job. + +Restarting +========== +A job stopped part way through a loop (killed, out of time) resumes in the +iteration it was in, at the step that had not finished, when rerun with +``--resume``; the JobServer does this for you. Finished iterations are not run +again. + +.. _parallel-iterations: + +Running iterations in parallel +============================== +With "Run iterations in parallel" set to yes, the iterations run at the same +time, each in an evaluator of its own, and what they did is brought back into +the job afterwards. This is worthwhile when each iteration does a substantial +calculation: starting an evaluator takes a few seconds, so for iterations of a +second or two a serial loop is faster. + +The contract +------------ +Choosing parallel declares that the iterations are independent: + +- Each iteration sees the job as it was when the loop started. An iteration + does not see what another one does, even an earlier one. +- What comes back, merged in iteration order: new and changed structures + (systems and configurations, with new ids), properties, table rows and + cells, new tables and columns, and files written in the job's directory + (``/name``). Files a step appends to, such as a structure file written by + Write Structure in append mode, get each iteration's part in order; other + files are those of the last iteration that wrote them, as for a serial loop; + a table saved inside the loop is saved again after the merge. +- Variables set in the body are not visible after the loop. +- ``break``, ``continue`` and skipping an iteration work as in a serial loop: + after a break in iteration *k* the iterations after *k* are not merged, so + the result is the serial one. +- A failed iteration is never merged (a serial loop keeps what a failed + iteration wrote before it failed). With "continue" it is reported and the + loop goes on; with "exit the loop" the iterations after it are not merged; + with "stop the job" the job fails, and resuming it runs the failed iteration + again. + +A loop whose iterations build on each other -- each reading a table row or a +structure an earlier iteration wrote, or overwriting the current structure that +an earlier iteration created -- gives a different result in parallel. Read +Structure's "Create a new system and configuration" keeps such iterations +independent. + +Settings +-------- +These appear when "Run iterations in parallel" is yes. + +Iterations at once + How many iterations run at the same time on this machine. By default as + many as the cores and memory allow, given the cores and memory per iteration. +Cores per iteration + The cores each iteration's calculations use (default 1). +Memory per iteration + The most memory an iteration needs (default 2 GB), so that the iterations + running together fit in the machine's memory. +Time per iteration + An estimate of an iteration's time, used to bundle iterations into queue + jobs when the job's calculations go to a queue. +Each iteration gets + *selected structures* (the default): the structure the iteration works on (a + loop over systems) and the current one, with all the tables. *Whole + database*: a copy of the whole database, for a body that reads other + structures. +Calculations run + *inline* (the default): each iteration's calculations run in its own share + of the machine. *Separate tasks*: they go to the job's target like any job's. +Two iterations writing one table cell + Two iterations setting the same cell of a row that existed before the loop + (or appending rows with the same index value) are an error by default; the + alternative keeps the later iteration's value, with a warning. + +Where things are +---------------- +Each iteration's step directories are where a serial loop puts them, +``//``. The iteration's own files -- its output +(``job.out``), its copy of the database and any job-level files it wrote -- are +in ``//_evaluator/``, which is kept for checking or debugging. + +Restarting a parallel loop +-------------------------- +A job stopped while its parallel loop runs resumes it: iterations already merged +are not run again, finished ones are merged, and running ones resume from their +own checkpoints. An iteration whose evaluator was killed while the job ran on +(out of memory, say) is run again, resuming where it was, up to twice. Index ===== diff --git a/loop_step/loop.py b/loop_step/loop.py index 1bf6bd8..2b7f2f0 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -35,6 +35,10 @@ def _plain(value): printer = printing.getPrinter("loop") +# How many times an iteration of a parallel loop whose evaluator stopped early +# (killed, out of memory) is run again, resuming from its checkpoint +MAX_RETRIES = 2 + def _yes(value): """A boolean parameter's value, as text or as a bool.""" @@ -957,35 +961,55 @@ def _run_parallel(self, P, context, checkpointer, state): ) ) - # Merge the iterations in order as they finish + # Merge the iterations in order as they finish. An iteration whose + # evaluator stopped before the iteration ended (killed, out of memory, + # a lost node) runs again, resuming from its own checkpoint. results = {} - runs = task_set.run() if n > 0 else iter(()) - try: - for result in runs: - results[keys[result.key]] = result - while state["next"] in results: - k = state["next"] - stopped = self._merge_iteration( - P, - context, - state, - merge_state, - k, - results.pop(k), - checkpointer, - system_db, - loop_directory, - job_directory, - later_wins, - ) + retries = {} + to_run = task_set if n > 0 else None + while to_run is not None: + again = [] + runs = to_run.run() + try: + for result in runs: + k = keys[result.key] + if self._stopped_early(state, k, loop_directory): + if retries.get(k, 0) < MAX_RETRIES: + retries[k] = retries.get(k, 0) + 1 + printer.job( + f" Loop iteration {state['directories'][str(k)]} " + "stopped before it ended; running it again." + ) + again.append(task_set.tasks[result.key]) + continue + results[k] = result + while state["next"] in results: + k = state["next"] + stopped = self._merge_iteration( + P, + context, + state, + merge_state, + k, + results.pop(k), + checkpointer, + system_db, + loop_directory, + job_directory, + later_wins, + ) + if stopped is not None: + break if stopped is not None: break - if stopped is not None: - break - finally: - # Cancels the iterations still running - if n > 0: + finally: + # Cancels the iterations still running runs.close() + to_run = None + if stopped is None and len(again) > 0: + to_run = TaskSet(self, local=pool, root=root) + for task in again: + to_run.add(task) if stopped is not None: k, why = stopped @@ -1034,6 +1058,30 @@ def _run_parallel(self, P, context, checkpointer, state): ) checkpointer.leave_loop(self) + def _evaluator_directory(self, state, k, loop_directory): + from seamm_exec import iteration as iterations + + return ( + loop_directory + / state["directories"][str(k)] + / iterations.EVALUATOR_DIRECTORY + ) + + def _stopped_early(self, state, k, loop_directory): + """Whether iteration ``k``'s evaluator stopped before the iteration ended. + + It did if its checkpoint is still 'running': an evaluator that ends, + with an error or not, says so in its checkpoint. + """ + path = self._evaluator_directory(state, k, loop_directory) / "seamm.db" + if not path.exists(): + return False + try: + checkpoint = seamm.read_checkpoint(path) + except Exception: + return False + return checkpoint is not None and checkpoint.get("state") == "running" + def _merge_iteration( self, P, @@ -1060,7 +1108,9 @@ def _merge_iteration( name = state["directories"][str(k)] iter_dir = loop_directory / name evaluator = iter_dir / iterations.EVALUATOR_DIRECTORY - outcome = iterations.iteration_outcome(evaluator) if result.ok else None + # The iteration's own final checkpoint says whether it ended: an + # evaluator killed as it exited, after finishing, did its work. + outcome = iterations.iteration_outcome(evaluator) if outcome is None: reason = result.reason or f"return code {result.returncode}" printer.job( From b7423f17ccbb33fa17e5b2b0b887e26a0c04fe50 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 18:25:02 -0400 Subject: [PATCH 3/6] Phase 6: a same-cell merge conflict names the Loop setting that allows it Co-Authored-By: Claude Opus 5.5 --- loop_step/loop.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/loop_step/loop.py b/loop_step/loop.py index 2b7f2f0..89ca018 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -1130,9 +1130,18 @@ def _merge_iteration( checkpointer.parallel_loop(self, state) return (k, "exit") - merged = iterations.merge_database( - system_db, evaluator, merge_state, k, later_wins=later_wins - ) + from molsystem.snapshot import MergeConflict + + try: + merged = iterations.merge_database( + system_db, evaluator, merge_state, k, later_wins=later_wins + ) + except MergeConflict as e: + text = str(e).replace(" (The Loop can let the later one win.)", "") + raise MergeConflict( + f"{text} Set the Loop's 'Two iterations writing one table cell' to " + "'the later iteration wins' to keep the later value instead." + ) from None if P["type"] == "For rows in table": # As a serial loop leaves it: the iteration's row, unless its body # moved on (below). The merge cannot see a move to the row that was From c0d49edf78339239f180b70d4582ccc8b449efd7 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 19:10:20 -0400 Subject: [PATCH 4/6] Phase 6 review: loop fixes and an end-to-end parallel test - A loop over systems makes the iteration's configuration current before merging it (M2). - Tables are re-exported and loop_entry.db removed after a break or exit too (S6); tables made in the body are variables after the merge (S7); merge warnings go to job.out; each iteration gets the chain of directories to read job-level files from. - tests/test_parallel.py runs a loop serially and in parallel with a fake step plug-in and compares the databases (S8). - Requires molsystem and seamm-exec; the user guide updated (conflicts on structures, appends, separate placement experimental, costs). Co-Authored-By: Claude Opus 5.5 --- docs/user_guide/index.rst | 31 +++--- loop_step/loop.py | 52 ++++++++-- requirements.txt | 2 + tests/test_parallel.py | 198 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 265 insertions(+), 18 deletions(-) create mode 100644 tests/test_parallel.py diff --git a/docs/user_guide/index.rst b/docs/user_guide/index.rst index 0dee7bc..cfd7e66 100644 --- a/docs/user_guide/index.rst +++ b/docs/user_guide/index.rst @@ -48,18 +48,22 @@ Running iterations in parallel With "Run iterations in parallel" set to yes, the iterations run at the same time, each in an evaluator of its own, and what they did is brought back into the job afterwards. This is worthwhile when each iteration does a substantial -calculation: starting an evaluator takes a few seconds, so for iterations of a -second or two a serial loop is faster. +calculation: starting an evaluator takes about five seconds, so for iterations +of a few seconds a serial loop is as fast or faster. The contract ------------ Choosing parallel declares that the iterations are independent: -- Each iteration sees the job as it was when the loop started. An iteration - does not see what another one does, even an earlier one. +- Each iteration sees the job's database as it was when the loop started. An + iteration does not see what another one does, even an earlier one. (A file in + the job's directory that an iteration reads, but has not written itself, is + read as it is at that moment.) - What comes back, merged in iteration order: new and changed structures - (systems and configurations, with new ids), properties, table rows and - cells, new tables and columns, and files written in the job's directory + (systems and configurations, with new ids), the current configuration of a + system, properties, table rows (appended after those of the iterations + before, as serially) and cells, new tables and columns, and files written in + the job's directory (``/name``). Files a step appends to, such as a structure file written by Write Structure in append mode, get each iteration's part in order; other files are those of the last iteration that wrote them, as for a serial loop; @@ -99,14 +103,19 @@ Each iteration gets *selected structures* (the default): the structure the iteration works on (a loop over systems) and the current one, with all the tables. *Whole database*: a copy of the whole database, for a body that reads other - structures. + structures. Every iteration's copy is written when the loop starts, so a + long loop over a large database needs that much disk space. Calculations run *inline* (the default): each iteration's calculations run in its own share - of the machine. *Separate tasks*: they go to the job's target like any job's. + of the machine. *Separate tasks* (experimental, not yet tested on a cluster): + they go to the job's target like any job's. Two iterations writing one table cell - Two iterations setting the same cell of a row that existed before the loop - (or appending rows with the same index value) are an error by default; the - alternative keeps the later iteration's value, with a warning. + Two iterations changing the same thing that existed before the loop -- a + cell of a table row, or a structure: its atoms' coordinates, velocities or + other values, or the structure itself -- are an error by default; the + alternative keeps the later iteration's change, with a warning in the + output. For example, a loop over methods that each optimize the current + structure in place is an error unless the later iteration may win. Where things are ---------------- diff --git a/loop_step/loop.py b/loop_step/loop.py index 89ca018..1107e15 100644 --- a/loop_step/loop.py +++ b/loop_step/loop.py @@ -40,6 +40,17 @@ def _plain(value): MAX_RETRIES = 2 +class _Collect(logging.Handler): + """Collects the warnings of a merge, for job.out.""" + + def __init__(self): + super().__init__(level=logging.WARNING) + self.messages = [] + + def emit(self, record): + self.messages.append(record.getMessage()) + + def _yes(value): """A boolean parameter's value, as text or as a bool.""" if isinstance(value, str): @@ -947,6 +958,12 @@ def _run_parallel(self, P, context, checkpointer, state): walltime=walltime, placement=placement, root_directory=self.flowchart.root_directory, + # A file the iteration has not written: this evaluator's, then + # those it reads itself + read_directories=[ + job_directory, + *getattr(self.flowchart, "job_read_directories", []), + ], ) task_set.add(task) keys[task.key] = k @@ -1029,13 +1046,15 @@ def _run_parallel(self, P, context, checkpointer, state): indent=self.indent + 4 * " ", ) ) - else: - # Tables the iterations exported, written again from the merged ones - for name, filename in state.get("exports", {}).items(): - try: - self.get_table(name, create=False).export(filename) - except Exception as e: - printer.important(f"Could not export the table '{name}': {e}") + # Tables the iterations exported, written again from the merged ones + for name, filename in state.get("exports", {}).items(): + try: + self.get_table(name, create=False).export(filename) + except Exception as e: + printer.important(f"Could not export the table '{name}': {e}") + if stopped is None or stopped[1] != "raise": + # Nothing more is dispatched (a resume after "stop the job" runs the + # failed iteration again, from the loop's entry) for suffix in ("", "-wal", "-shm"): Path(str(entry) + suffix).unlink(missing_ok=True) @@ -1132,6 +1151,17 @@ def _merge_iteration( from molsystem.snapshot import MergeConflict + if P["type"] == "For systems in the database": + # As a serial loop leaves it: the iteration's configuration current, + # unless its body chose another (merged below). The merge cannot see + # a choice of the configuration that was current at the loop's entry. + configuration = context["configurations"][k - 1] + system_db.system = configuration.system + configuration.system.configuration = configuration + tables_before = set(system_db.user_tables) + warnings = _Collect() + merge_logger = logging.getLogger("molsystem.snapshot") + merge_logger.addHandler(warnings) try: merged = iterations.merge_database( system_db, evaluator, merge_state, k, later_wins=later_wins @@ -1142,6 +1172,14 @@ def _merge_iteration( f"{text} Set the Loop's 'Two iterations writing one table cell' to " "'the later iteration wins' to keep the later value instead." ) from None + finally: + merge_logger.removeHandler(warnings) + for message in warnings.messages: + printer.job(f" Warning: {message}") + # Tables made in the body are variables, as a serial loop leaves them + for table in set(system_db.user_tables) - tables_before: + if not self.variable_exists(table): + self.set_variable(table, seamm.Table(system_db, table)) if P["type"] == "For rows in table": # As a serial loop leaves it: the iteration's row, unless its body # moved on (below). The merge cannot see a move to the row that was diff --git a/requirements.txt b/requirements.txt index 311ea9b..fff5099 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ +molsystem>=2026.10.4 psutil seamm>=2026.10.4 +seamm-exec>=2026.10.4.1 diff --git a/tests/test_parallel.py b/tests/test_parallel.py new file mode 100644 index 0000000..4faec53 --- /dev/null +++ b/tests/test_parallel.py @@ -0,0 +1,198 @@ +# -*- coding: utf-8 -*- + +"""A parallel Loop, end to end, without external codes (phase 6). + +A tiny step plug-in, installed for the test on PYTHONPATH (a module and its +dist-info with the entry points), makes a system named after the loop's value, +appends a row to a table it creates if needed, and writes a job-level file. The +same flowchart is run serially and in parallel, each by ``run_flowchart`` in its +own directory, and the databases and files are compared. +""" + +import os +from pathlib import Path +import sqlite3 +import subprocess +import sys +import textwrap + +import pytest + +BIN = Path(sys.executable).parent + +FAKE_STEP = textwrap.dedent(''' + """A step for testing parallel loops: a system, a table row, a file.""" + + import seamm + + + class FakeParameters(seamm.Parameters): + parameters = { + "name": { + "default": "x", + "kind": "string", + "default_units": "", + "enumeration": tuple(), + "format_string": "s", + "description": "Name:", + "help_text": "The name of the system and row.", + }, + } + + def __init__(self, defaults={}, data=None): + super().__init__( + defaults={**FakeParameters.parameters, **defaults}, data=data + ) + + + class Fake(seamm.Node): + def __init__(self, flowchart=None, extension=None): + super().__init__(flowchart=flowchart, title="Fake", extension=extension) + self.parameters = FakeParameters() + + @property + def version(self): + return "0.1" + + def description_text(self, P=None): + return self.header + "\\n A fake step." + + def run(self): + next_node = super().run(None) + P = self.parameters.current_values_to_dict( + context=seamm.flowchart_variables._data + ) + name = str(P["name"]) + db = self.get_variable("_system_db") + system = db.create_system(name=name) + configuration = system.create_configuration(name=name) + configuration.atoms.append( + x=[0.0], y=[0.0], z=[float(len(name))], symbol=["He"] + ) + db.system = system + table = self.get_table("results") + if "name" not in table.columns: + table.add_column("name", "string", "") + table.append_row(name=name) + (self.job_path / "last.txt").write_text(name + "\\n") + return next_node + + + class FakeStep: + my_description = { + "description": "A fake step for tests", + "group": "Control", + "name": "Fake", + } + + def __init__(self, flowchart=None, gui=None): + pass + + def description(self): + return FakeStep.my_description + + def create_node(self, flowchart=None, **kwargs): + return Fake(flowchart=flowchart, **kwargs) + + def create_tk_node(self, canvas=None, **kwargs): + raise NotImplementedError("no GUI for the fake step") +''') + +SPEC = textwrap.dedent("""\ + title: Parallel loop test + steps: + - Loop: + type: Foreach + variable: name + values: alpha beta gamma delta + {parallel} + body: + - Fake: + name: $name + """) + + +@pytest.fixture(scope="module") +def fake_plugin(tmp_path_factory): + """A site directory holding the fake step and its entry points.""" + site = tmp_path_factory.mktemp("site") + (site / "fakestep.py").write_text(FAKE_STEP) + info = site / "fakestep-0.1.dist-info" + info.mkdir() + (info / "METADATA").write_text( + "Metadata-Version: 2.1\nName: fakestep\nVersion: 0.1\n" + ) + (info / "entry_points.txt").write_text( + "[org.molssi.seamm]\nFake = fakestep:FakeStep\n\n" + "[org.molssi.seamm.tk]\nFake = fakestep:FakeStep\n" + ) + return site + + +def environment(site): + env = { + k: v + for k, v in os.environ.items() + if not k.startswith(("SEAMM_RESUME", "SEAMM_PARENT_JOB", "SEAMM_CE")) + } + env["PYTHONPATH"] = os.pathsep.join( + [str(site), *filter(None, [os.environ.get("PYTHONPATH")])] + ) + return env + + +def run(site, directory, parallel): + directory.mkdir() + text = "parallel: 'yes'" if parallel else "" + if parallel: + text += "\n iterations at once: '2'\n memory per iteration: '0.2'" + (directory / "spec.yaml").write_text(SPEC.format(parallel=text)) + env = environment(site) + subprocess.run( + [str(BIN / "seamm-flowchart"), "build", "spec.yaml", "-o", "test.flow"], + cwd=directory, + env=env, + check=True, + capture_output=True, + ) + result = subprocess.run( + [str(BIN / "run_flowchart"), "test.flow"], + cwd=directory, + env=env, + capture_output=True, + text=True, + timeout=600, + ) + assert result.returncode == 0, result.stdout[-3000:] + result.stderr[-3000:] + return directory + + +def contents(directory): + db = sqlite3.connect(f"file:{directory / 'seamm.db'}?mode=ro", uri=True) + try: + systems = db.execute("SELECT name FROM system ORDER BY id").fetchall() + coordinates = db.execute( + "SELECT configuration, z FROM coordinates ORDER BY configuration" + ).fetchall() + (sql_name,) = db.execute( + "SELECT sql_name FROM _tables WHERE name = 'results'" + ).fetchone() + rows = db.execute(f'SELECT * FROM "{sql_name}"').fetchall() + finally: + db.close() + return systems, coordinates, rows, (directory / "last.txt").read_text() + + +@pytest.mark.skipif( + not (BIN / "run_flowchart").exists() or not (BIN / "seamm-flowchart").exists(), + reason="run_flowchart and seamm-flowchart are not installed beside Python", +) +def test_parallel_loop_gives_the_serial_result(fake_plugin, tmp_path): + serial = run(fake_plugin, tmp_path / "serial", parallel=False) + parallel = run(fake_plugin, tmp_path / "parallel", parallel=True) + expected = contents(serial) + assert [s for (s,) in expected[0]] == ["alpha", "beta", "gamma", "delta"] + assert contents(parallel) == expected + # Each iteration ran in an evaluator of its own + evaluators = sorted(p.parent.name for p in parallel.rglob("_evaluator/job.out")) + assert len(evaluators) == 4 From 7266fc3d775998a855c8302be89f0bce576b8b2f Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Sun, 4 Oct 2026 19:16:01 -0400 Subject: [PATCH 5/6] Phase 6 test: build and run with the source under test The end-to-end parallel test builds and runs its flowcharts with this interpreter and the package's source first on PYTHONPATH (inherited by the iterations), not the installed console scripts; failures show their output. Co-Authored-By: Claude Opus 5.5 --- tests/test_parallel.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/tests/test_parallel.py b/tests/test_parallel.py index 4faec53..b8d4cb3 100644 --- a/tests/test_parallel.py +++ b/tests/test_parallel.py @@ -19,6 +19,10 @@ import pytest BIN = Path(sys.executable).parent +# The source under test first, for this process's tools and the iterations' +SOURCE = Path(__file__).resolve().parents[1] +BUILD = "import sys; from seamm.flowchart_cli import main; sys.exit(main())" +RUN = "import sys; from seamm_exec import run; sys.argv[0] = 'run_flowchart'; run()" FAKE_STEP = textwrap.dedent(''' """A step for testing parallel loops: a system, a table row, a file.""" @@ -136,7 +140,7 @@ def environment(site): if not k.startswith(("SEAMM_RESUME", "SEAMM_PARENT_JOB", "SEAMM_CE")) } env["PYTHONPATH"] = os.pathsep.join( - [str(site), *filter(None, [os.environ.get("PYTHONPATH")])] + [str(SOURCE), str(site), *filter(None, [os.environ.get("PYTHONPATH")])] ) return env @@ -148,15 +152,16 @@ def run(site, directory, parallel): text += "\n iterations at once: '2'\n memory per iteration: '0.2'" (directory / "spec.yaml").write_text(SPEC.format(parallel=text)) env = environment(site) - subprocess.run( - [str(BIN / "seamm-flowchart"), "build", "spec.yaml", "-o", "test.flow"], + built = subprocess.run( + [sys.executable, "-c", BUILD, "build", "spec.yaml", "-o", "test.flow"], cwd=directory, env=env, - check=True, capture_output=True, + text=True, ) + assert built.returncode == 0, built.stdout[-3000:] + built.stderr[-3000:] result = subprocess.run( - [str(BIN / "run_flowchart"), "test.flow"], + [sys.executable, "-c", RUN, "test.flow"], cwd=directory, env=env, capture_output=True, @@ -184,8 +189,8 @@ def contents(directory): @pytest.mark.skipif( - not (BIN / "run_flowchart").exists() or not (BIN / "seamm-flowchart").exists(), - reason="run_flowchart and seamm-flowchart are not installed beside Python", + not (BIN / "run_flowchart").exists(), + reason="run_flowchart (for the iterations) is not installed beside Python", ) def test_parallel_loop_gives_the_serial_result(fake_plugin, tmp_path): serial = run(fake_plugin, tmp_path / "serial", parallel=False) From 096fde0b63c0d33fb323ad4ca563fed87df7da57 Mon Sep 17 00:00:00 2001 From: Paul Saxe Date: Mon, 5 Oct 2026 06:16:30 -0400 Subject: [PATCH 6/6] Release 2026.10.5: HISTORY, pins and documentation (phase 6) Co-Authored-By: Claude Opus 5.5 --- HISTORY.rst | 24 ++++++++++++++++++++++++ requirements.txt | 6 +++--- setup.py | 2 +- 3 files changed, 28 insertions(+), 4 deletions(-) diff --git a/HISTORY.rst b/HISTORY.rst index bca2dac..5826239 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -1,6 +1,30 @@ ======= History ======= +2026.10.5 -- Run a loop's iterations in parallel + * New option "Run iterations in parallel" (off by default). Each iteration then + runs in an evaluator of its own, at the same time as others, on this machine or + bundled into queue jobs, and what it did -- structures, properties, table rows + and cells, files written in the job's directory -- is merged back in iteration + order. The iterations must be independent: each sees the job's database as it + was when the loop started (a job-level file it reads is read as it is at that + moment), and variables set in the loop are not visible after it. ``break``, + ``continue`` and skipping work as before; a failed iteration is never merged. + Two iterations changing the same table cell or structure are an error unless + "the later iteration wins" is chosen, which keeps the later change and writes a + warning to ``job.out``. + * Its settings, shown only for a parallel loop: iterations at once (by default + as many as cores and memory allow), cores, memory and time per iteration, what + each iteration gets (the structures it works on, or the whole database -- every + iteration's copy is written as the loop starts, which needs disk space), where + calculations run (inline, or as separate tasks -- experimental) and same-cell + writes. Starting an evaluator takes about five seconds, so short iterations are + better run serially. + * A job stopped part way through a parallel loop resumes it: merged iterations are + not run again and running ones resume where they were; an iteration whose + evaluator was killed is run again, up to twice. + * A user guide for the Loop step. + * Requires molsystem, seamm and seamm-exec 2026.10.5. 2026.10.4 -- Loops resume where they stopped * A job resumed from its checkpoint (seamm_exec 2026.10.4.1, ``--resume``) continues a loop at the iteration and step it had reached, with the same diff --git a/requirements.txt b/requirements.txt index fff5099..9f4d8a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,4 @@ -molsystem>=2026.10.4 +molsystem>=2026.10.5 psutil -seamm>=2026.10.4 -seamm-exec>=2026.10.4.1 +seamm>=2026.10.5 +seamm-exec>=2026.10.5 diff --git a/setup.py b/setup.py index 99c3187..8ffdcf3 100644 --- a/setup.py +++ b/setup.py @@ -54,7 +54,7 @@ install_requires=requirements, # The resume tests run flowcharts through the evaluator - extras_require={"test": ["seamm-exec>=2026.10.4.1"]}, + extras_require={"test": ["seamm-exec>=2026.10.5"]}, test_suite='tests',