Enable multiprocessing groups within project config

spacy/cli/_util.py

@@ -1,5 +1,5 @@
 from typing import Dict, Any, Union, List, Optional, Tuple, Iterable
-from typing import TYPE_CHECKING, overload
+from typing import TYPE_CHECKING, overload, cast
 import sys
 import shutil
 from pathlib import Path
@@ -18,8 +18,12 @@
 
 from ..compat import Literal
 from ..schemas import ProjectConfigSchema, validate
+from ..git_info import GIT_VERSION
 from ..util import import_file, run_command, make_tempdir, registry, logger
 from ..util import is_compatible_version, SimpleFrozenDict, ENV_VARS
+from ..util import is_minor_version_match
+
+
 from .. import about
 
 if TYPE_CHECKING:
@@ -157,6 +161,7 @@ def load_project_config(
         print("\n".join(errors))
         sys.exit(1)
     validate_project_version(config)
+    validate_max_parallel_processes(config)
     validate_project_commands(config)
     # Make sure directories defined in config exist
     for subdir in config.get("directories", []):
@@ -199,7 +204,7 @@ def substitute_project_variables(
 
 
 def validate_project_version(config: Dict[str, Any]) -> None:
-    """If the project defines a compatible spaCy version range, chec that it's
+    """If the project defines a compatible spaCy version range, check that it's
     compatible with the current version of spaCy.
 
     config (Dict[str, Any]): The loaded config.
@@ -215,30 +220,67 @@ def validate_project_version(config: Dict[str, Any]) -> None:
         msg.fail(err, exits=1)
 
 
+def validate_max_parallel_processes(config: Dict[str, Any]) -> None:
+    """If the project defines a maximum number of parallel processes, check that the
+    value is within the permitted range.
+
+    config (Dict[str, Any]): The loaded config.
+    """
+    max_parallel_processes = config.get("max_parallel_processes", None)
+    if max_parallel_processes is not None and max_parallel_processes < 2:
+        err = (
+            f"The {PROJECT_FILE} specifies a value for max_parallel_processes ({max_parallel_processes}) "
+            f"that is less than 2."
+        )
+        msg.fail(err, exits=1)
+
+
+def verify_workflow_step(workflow_name: str, commands: List[str], step: str) -> None:
+    if step not in commands:
+        msg.fail(
+            f"Unknown command specified in workflow '{workflow_name}': {step}",
+            f"Workflows can only refer to commands defined in the 'commands' "
+            f"section of the {PROJECT_FILE}.",
+            exits=1,
+        )
+
+
 def validate_project_commands(config: Dict[str, Any]) -> None:
     """Check that project commands and workflows are valid, don't contain
-    duplicates, don't clash  and only refer to commands that exist.
+    duplicates, don't clash and only refer to commands that exist.
 
     config (Dict[str, Any]): The loaded config.
     """
-    command_names = [cmd["name"] for cmd in config.get("commands", [])]
+
+    commands = [cmd["name"] for cmd in config.get("commands", [])]
     workflows = config.get("workflows", {})
-    duplicates = set([cmd for cmd in command_names if command_names.count(cmd) > 1])
+    duplicates = set([cmd for cmd in commands if commands.count(cmd) > 1])
     if duplicates:
         err = f"Duplicate commands defined in {PROJECT_FILE}: {', '.join(duplicates)}"
         msg.fail(err, exits=1)
-    for workflow_name, workflow_steps in workflows.items():
-        if workflow_name in command_names:
+    for workflow_name, workflow_items in workflows.items():
+        if workflow_name in commands:
             err = f"Can't use workflow name '{workflow_name}': name already exists as a command"
             msg.fail(err, exits=1)
-        for step in workflow_steps:
-            if step not in command_names:
-                msg.fail(
-                    f"Unknown command specified in workflow '{workflow_name}': {step}",
-                    f"Workflows can only refer to commands defined in the 'commands' "
-                    f"section of the {PROJECT_FILE}.",
-                    exits=1,
-                )
+        for workflow_item in workflow_items:
+            if isinstance(workflow_item, str):
+                verify_workflow_step(workflow_name, commands, workflow_item)
+            else:
+                steps = cast(List[str], workflow_item["parallel"])
+                if len(steps) < 2:
+                    msg.fail(
+                        f"Invalid parallel group within '{workflow_name}'.",
+                        f"A parallel group must reference at least two commands.",
+                        exits=1,
+                    )
+                if len(steps) != len(set(steps)):
+                    msg.fail(
+                        f"Invalid parallel group within '{workflow_name}'.",
+                        f"A parallel group may not contain a command more than once.",
+                        exits=1,
+                    )
+                for step in steps:
+                    verify_workflow_step(workflow_name, commands, step)
 
 
 def get_hash(data, exclude: Iterable[str] = tuple()) -> str:
@@ -573,3 +615,121 @@ def setup_gpu(use_gpu: int, silent=None) -> None:
         local_msg.info("Using CPU")
         if gpu_is_available():
             local_msg.info("To switch to GPU 0, use the option: --gpu-id 0")
+
+
+def check_rerun(
+    project_dir: Path,
+    command: Dict[str, Any],
+    *,
+    check_spacy_version: bool = True,
+    check_spacy_commit: bool = False,
+) -> bool:
+    """Check if a command should be rerun because its settings or inputs/outputs
+    changed.
+
+    project_dir (Path): The current project directory.
+    command (Dict[str, Any]): The command, as defined in the project.yml.
+    strict_version (bool):
+    RETURNS (bool): Whether to re-run the command.
+    """
+    # Always rerun if no-skip is set
+    if command.get("no_skip", False):
+        return True
+    lock_path = project_dir / PROJECT_LOCK
+    if not lock_path.exists():  # We don't have a lockfile, run command
+        return True
+    data = srsly.read_yaml(lock_path)
+    if command["name"] not in data:  # We don't have info about this command
+        return True
+    entry = data[command["name"]]
+    # Always run commands with no outputs (otherwise they'd always be skipped)
+    if not entry.get("outs", []):
+        return True
+    # Always rerun if spaCy version or commit hash changed
+    spacy_v = entry.get("spacy_version")
+    commit = entry.get("spacy_git_version")
+    if check_spacy_version and not is_minor_version_match(spacy_v, about.__version__):
+        info = f"({spacy_v} in {PROJECT_LOCK}, {about.__version__} current)"
+        msg.info(f"Re-running '{command['name']}': spaCy minor version changed {info}")
+        return True
+    if check_spacy_commit and commit != GIT_VERSION:
+        info = f"({commit} in {PROJECT_LOCK}, {GIT_VERSION} current)"
+        msg.info(f"Re-running '{command['name']}': spaCy commit changed {info}")
+        return True
+    # If the entry in the lockfile matches the lockfile entry that would be
+    # generated from the current command, we don't rerun because it means that
+    # all inputs/outputs, hashes and scripts are the same and nothing changed
+    lock_entry = _get_lock_entry(project_dir, command)
+    exclude = ["spacy_version", "spacy_git_version"]
+    return get_hash(lock_entry, exclude=exclude) != get_hash(entry, exclude=exclude)
+
+
+def update_lockfile(
+    project_dir: Path,
+    command: Dict[str, Any],
+) -> None:
+    """Update the lockfile after running a command. Will create a lockfile if
+    it doesn't yet exist and will add an entry for the current command, its
+    script and dependencies/outputs.
+
+    project_dir (Path): The current project directory.
+    command (Dict[str, Any]): The command, as defined in the project.yml.
+    """
+    lock_path = project_dir / PROJECT_LOCK
+    if not lock_path.exists():
+        srsly.write_yaml(lock_path, {})
+        data = {}
+    else:
+        data = srsly.read_yaml(lock_path)
+    data[command["name"]] = _get_lock_entry(project_dir, command)
+    srsly.write_yaml(lock_path, data)
+
+
+def check_deps(cmd: Dict, cmd_name: str, project_dir: Path, dry: bool):
+    for dep in cmd.get("deps", []):
+        if not (project_dir / dep).exists():
+            err = f"Missing dependency specified by command '{cmd_name}': {dep}"
+            err_help = "Maybe you forgot to run the 'project assets' command or a previous step?"
+            err_kwargs = {"exits": 1} if not dry else {}
+            msg.fail(err, err_help, **err_kwargs)
+
+
+def _get_lock_entry(project_dir: Path, command: Dict[str, Any]) -> Dict[str, Any]:
+    """Get a lockfile entry for a given command. An entry includes the command,
+    the script (command steps) and a list of dependencies and outputs with
+    their paths and file hashes, if available. The format is based on the
+    dvc.lock files, to keep things consistent.
+
+    project_dir (Path): The current project directory.
+    command (Dict[str, Any]): The command, as defined in the project.yml.
+    RETURNS (Dict[str, Any]): The lockfile entry.
+    """
+    deps = _get_fileinfo(project_dir, command.get("deps", []))
+    outs = _get_fileinfo(project_dir, command.get("outputs", []))
+    outs_nc = _get_fileinfo(project_dir, command.get("outputs_no_cache", []))
+    return {
+        "cmd": f"{COMMAND} run {command['name']}",
+        "script": command["script"],
+        "deps": deps,
+        "outs": [*outs, *outs_nc],
+        "spacy_version": about.__version__,
+        "spacy_git_version": GIT_VERSION,
+    }
+
+
+def _get_fileinfo(
+    project_dir: Path, paths: List[str]
+) -> List[Dict[str, Optional[str]]]:
+    """Generate the file information for a list of paths (dependencies, outputs).
+    Includes the file path and the file's checksum.
+
+    project_dir (Path): The current project directory.
+    paths (List[str]): The file paths.
+    RETURNS (List[Dict[str, str]]): The lockfile entry for a file.
+    """
+    data = []
+    for path in paths:
+        file_path = project_dir / path
+        md5 = get_checksum(file_path) if file_path.exists() else None
+        data.append({"path": path, "md5": md5})
+    return data

spacy/cli/project/document.py

@@ -14,8 +14,9 @@
 Commands are only re-run if their inputs have changed."""
 INTRO_WORKFLOWS = f"""The following workflows are defined by the project. They
 can be executed using [`spacy project run [name]`]({DOCS_URL}/api/cli#project-run)
-and will run the specified commands in order. Commands are only re-run if their
-inputs have changed."""
+and will run the specified commands in order. Commands grouped within square brackets
+are run in parallel. Commands are only re-run if their inputs 
+have changed."""
 INTRO_ASSETS = f"""The following assets are defined by the project. They can
 be fetched by running [`spacy project assets`]({DOCS_URL}/api/cli#project-assets)
 in the project directory."""
@@ -69,7 +70,19 @@ def project_document(
         md.add(md.table(data, ["Command", "Description"]))
     # Workflows
     wfs = config.get("workflows", {}).items()
-    data = [(md.code(n), " → ".join(md.code(w) for w in stp)) for n, stp in wfs]
+    data = []
+    for n, steps in wfs:
+        rendered_steps = []
+        for step in steps:
+            if isinstance(step, str):
+                rendered_steps.append(md.code(step))
+            else:
+                rendered_steps.append(
+                    "["
+                    + ", ".join(md.code(p_step) for p_step in step["parallel"])
+                    + "]"
+                )
+        data.append([md.code(n), " → ".join(rendered_steps)])
     if data:
         md.add(md.title(3, "Workflows", "⏭"))
         md.add(INTRO_WORKFLOWS)

spacy/cli/project/dvc.py

@@ -1,5 +1,5 @@
 """This module contains helpers and subcommands for integrating spaCy projects
-with Data Version Controk (DVC). https://dvc.org"""
+with Data Version Control (DVC). https://dvc.org"""
 from typing import Dict, Any, List, Optional, Iterable
 import subprocess
 from pathlib import Path
@@ -106,6 +106,11 @@ def update_dvc_config(
     dvc_commands = []
     config_commands = {cmd["name"]: cmd for cmd in config.get("commands", [])}
     for name in workflows[workflow]:
+        if isinstance(name, dict) and "parallel" in name:
+            msg.fail(
+                f"A DVC workflow may not contain parallel groups",
+                exits=1,
+            )
         command = config_commands[name]
         deps = command.get("deps", [])
         outputs = command.get("outputs", [])
@@ -123,6 +128,11 @@ def update_dvc_config(
             dvc_cmd.append("--always-changed")
         full_cmd = [*dvc_cmd, *deps_cmd, *outputs_cmd, *outputs_nc_cmd, *project_cmd]
         dvc_commands.append(join_command(full_cmd))
+    if len(dvc_commands) == 0:
+        msg.fail(
+            f"A DVC workflow must have at least one dependency or output",
+            exits=1,
+        )
     with working_dir(path):
         dvc_flags = {"--verbose": verbose, "--quiet": silent}
         run_dvc_commands(dvc_commands, flags=dvc_flags)