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/docs/user_guide/index.rst b/docs/user_guide/index.rst index ef780d3..cfd7e66 100644 --- a/docs/user_guide/index.rst +++ b/docs/user_guide/index.rst @@ -3,14 +3,133 @@ ********** 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 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'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), 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; + 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. 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* (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 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 +---------------- +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 014beb5..1107e15 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 @@ -32,6 +35,44 @@ 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 + + +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): + 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.""" @@ -236,18 +277,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 +382,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 +433,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 +713,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 +737,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 +753,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 +779,455 @@ 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, + # 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 + + 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. An iteration whose + # evaluator stopped before the iteration ended (killed, out of memory, + # a lost node) runs again, resuming from its own checkpoint. + results = {} + 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 + 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 + 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 * " ", + ) + ) + # 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) + + 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 _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, + 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 + # 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( + 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") + + 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 + ) + 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 + 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 + # 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/requirements.txt b/requirements.txt index 311ea9b..9f4d8a9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ +molsystem>=2026.10.5 psutil -seamm>=2026.10.4 +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', 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() diff --git a/tests/test_parallel.py b/tests/test_parallel.py new file mode 100644 index 0000000..b8d4cb3 --- /dev/null +++ b/tests/test_parallel.py @@ -0,0 +1,203 @@ +# -*- 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 +# 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.""" + + 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(SOURCE), 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) + built = subprocess.run( + [sys.executable, "-c", BUILD, "build", "spec.yaml", "-o", "test.flow"], + cwd=directory, + env=env, + capture_output=True, + text=True, + ) + assert built.returncode == 0, built.stdout[-3000:] + built.stderr[-3000:] + result = subprocess.run( + [sys.executable, "-c", RUN, "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(), + 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) + 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