From 08300d3bcf51f7bd9c8509924110416461b11b60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Peter=20Rowlands=20=28=EB=B3=80=EA=B8=B0=ED=98=B8=29?= Date: Fri, 29 Jan 2021 13:18:20 +0900 Subject: [PATCH] experiments: drop `exp resume` in favor of auto-resuming in `exp run` (#5351) * exp: auto determine checkpoint resume rev based on workspace state * exp: add 'dvc exp reset' for reset/restarting checkpoints * update tests * fix reset usage --- dvc/command/experiments.py | 56 +++++--- dvc/repo/experiments/__init__.py | 151 +++++++++++---------- dvc/repo/experiments/run.py | 11 +- tests/func/experiments/conftest.py | 1 - tests/func/experiments/test_checkpoints.py | 55 +------- tests/unit/command/test_experiments.py | 8 +- 6 files changed, 131 insertions(+), 151 deletions(-) diff --git a/dvc/command/experiments.py b/dvc/command/experiments.py index 1087fa414..802d206bf 100644 --- a/dvc/command/experiments.py +++ b/dvc/command/experiments.py @@ -471,6 +471,9 @@ class CmdExperimentsDiff(CmdBase): class CmdExperimentsRun(CmdRepro): def run(self): + if self.args.reset: + logger.info("Any existing checkpoints will be reset and re-run.") + self.repo.experiments.run( name=self.args.name, queue=self.args.queue, @@ -478,6 +481,7 @@ class CmdExperimentsRun(CmdRepro): jobs=self.args.jobs, params=self.args.params, checkpoint_resume=self.args.checkpoint_resume, + reset=self.args.reset, tmp_dir=self.args.tmp_dir, **self._repro_kwargs, ) @@ -629,7 +633,6 @@ class CmdExperimentsPull(CmdBase): def add_parser(subparsers, parent_parser): EXPERIMENTS_HELP = "Commands to run and compare experiments." - LAST_CHECKPOINT = ":last" experiments_parser = subparsers.add_parser( "experiments", @@ -842,34 +845,47 @@ def add_parser(subparsers, parent_parser): ) _add_run_common(experiments_run_parser) experiments_run_parser.add_argument( - "--checkpoint-resume", type=str, default=None, help=argparse.SUPPRESS, - ) - experiments_run_parser.set_defaults(func=CmdExperimentsRun) - - EXPERIMENTS_RESUME_HELP = "Resume checkpoint experiments." - experiments_resume_parser = experiments_subparsers.add_parser( - "resume", - parents=[parent_parser], - aliases=["res"], - description=append_doc_link(EXPERIMENTS_RESUME_HELP, "exp/resume"), - help=EXPERIMENTS_RESUME_HELP, - formatter_class=argparse.RawDescriptionHelpFormatter, - ) - _add_run_common(experiments_resume_parser) - experiments_resume_parser.add_argument( "-r", "--rev", type=str, - default=LAST_CHECKPOINT, dest="checkpoint_resume", help=( "Continue the specified checkpoint experiment. " - "If no experiment revision is provided, " - "the most recently run checkpoint experiment will be used." + "(Only required for explicitly resuming checkpoints in queued " + "or temp dir runs.)" ), metavar="", ) - experiments_resume_parser.set_defaults(func=CmdExperimentsRun) + experiments_run_parser.add_argument( + "--reset", action="store_true", help=argparse.SUPPRESS, + ) + experiments_run_parser.set_defaults(func=CmdExperimentsRun) + + EXPERIMENTS_RESET_HELP = "Reset and restart checkpoint experiments." + experiments_reset_parser = experiments_subparsers.add_parser( + "reset", + parents=[parent_parser], + description=append_doc_link(EXPERIMENTS_RESET_HELP, "exp/reset"), + help=EXPERIMENTS_RESET_HELP, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + experiments_reset_parser.add_argument( + "-r", + "--rev", + type=str, + default=None, + dest="checkpoint_resume", + help=argparse.SUPPRESS, + ) + _add_run_common(experiments_reset_parser) + experiments_reset_parser.add_argument( + "--reset", + action="store_const", + const=True, + default=True, + help=argparse.SUPPRESS, + ) + experiments_reset_parser.set_defaults(func=CmdExperimentsRun) EXPERIMENTS_GC_HELP = "Garbage collect unneeded experiments." EXPERIMENTS_GC_DESCRIPTION = ( diff --git a/dvc/repo/experiments/__init__.py b/dvc/repo/experiments/__init__.py index 15ae734d9..1c7723927 100644 --- a/dvc/repo/experiments/__init__.py +++ b/dvc/repo/experiments/__init__.py @@ -82,7 +82,6 @@ class Experiments: r"^(?P[a-f0-9]{7})-(?P[a-f0-9]+)" r"(?P-checkpoint)?$" ) - LAST_CHECKPOINT = ":last" EXEC_TMP_DIR = "exps" StashEntry = namedtuple( @@ -192,23 +191,30 @@ class Experiments: self._update_params(params) if resume_rev: + if branch: + branch_name = ExpRefInfo.from_ref(branch).name + else: + branch_name = "" if self.scm.is_dirty(): - logger.debug( - "Resume new checkpoint branch with " - "modifications" + logger.info( + "Modified checkpoint experiment based on " + "'%s' will be created", + branch_name, ) branch = None else: - logger.debug( - "Resuming from tip of existing branch '%s'", - branch, + logger.info( + "Existing checkpoint experiment '%s' will be " + "resumed", + branch_name, + ) + if name: + logger.warning( + "Ignoring option '--name %s' for resumed " + "experiment. Existing experiment name will" + "be preserved instead.", + name, ) - if name: - logger.warning( - "Ignoring option '--name %s' for resumed " - "experiment. Existing experiment name will" - "be preserved instead." - ) # save additional repro command line arguments run_env = {DVCLIVE_RESUME: "1"} if resume_rev else {} @@ -323,7 +329,13 @@ class Experiments: # whether the file is dirty self.scm.add(list(params.keys())) - def reproduce_one(self, queue=False, tmp_dir=False, **kwargs): + def reproduce_one( + self, + queue: bool = False, + tmp_dir: bool = False, + checkpoint_resume: Optional[str] = None, + **kwargs, + ): """Reproduce and checkout a single experiment.""" if not (queue or tmp_dir): staged, _, _ = self.scm.status() @@ -334,7 +346,20 @@ class Experiments: ) self.scm.reset() - stash_rev = self.new(**kwargs) + if checkpoint_resume: + resume_rev = self.scm.resolve_rev(checkpoint_resume) + try: + self.check_baseline(resume_rev) + checkpoint_resume = resume_rev + except BaselineMismatchError as exc: + raise DvcException( + f"Cannot resume from '{checkpoint_resume}' as it is not " + "derived from your current workspace." + ) from exc + else: + checkpoint_resume = self._workspace_resume_rev() + + stash_rev = self.new(checkpoint_resume=checkpoint_resume, **kwargs) if queue: logger.info( "Queued experiment '%s' for future execution.", stash_rev[:7], @@ -349,6 +374,13 @@ class Experiments: self._log_reproduced(results, tmp_dir=tmp_dir) return results + def _workspace_resume_rev(self) -> Optional[str]: + last_checkpoint = self._get_last_checkpoint() + last_applied = self._get_last_applied() + if last_checkpoint and last_applied: + return last_applied + return None + def reproduce_queued(self, **kwargs): results = self._reproduce_revs(**kwargs) if results: @@ -387,26 +419,20 @@ class Experiments: """ if checkpoint_resume is not None: return self._resume_checkpoint( - *args, checkpoint_resume=checkpoint_resume, **kwargs + *args, resume_rev=checkpoint_resume, **kwargs ) return self._stash_exp(*args, **kwargs) def _resume_checkpoint( - self, *args, checkpoint_resume: Optional[str] = None, **kwargs, + self, *args, resume_rev: Optional[str] = None, **kwargs, ): """Resume an existing (checkpoint) experiment. Experiment will be reproduced and checked out into the user's workspace. """ - assert checkpoint_resume - - if checkpoint_resume == self.LAST_CHECKPOINT: - # Continue from most recently committed checkpoint - resume_rev = self._get_last_checkpoint() - else: - resume_rev = self.scm.resolve_rev(checkpoint_resume) + assert resume_rev allow_multiple = "params" in kwargs branch: Optional[str] = self.get_branch_by_rev( @@ -414,54 +440,13 @@ class Experiments: ) if not branch: raise DvcException( - "Could not find checkpoint experiment " - f"'{checkpoint_resume}'" + "Could not find checkpoint experiment " f"'{resume_rev[:7]}'" ) - last_applied = self.scm.get_ref(EXEC_APPLY) - try: - if last_applied: - self.check_baseline(last_applied) - self.check_baseline(resume_rev) - except BaselineMismatchError: - # If HEAD has moved since the the last applied checkpoint, - # the applied checkpoint is no longer valid - self.scm.remove_ref(EXEC_APPLY) - last_applied = None - checkpoint_resume = None - if resume_rev != last_applied: - if checkpoint_resume == self.LAST_CHECKPOINT: - display_rev: Optional[str] = resume_rev[:7] - else: - display_rev = checkpoint_resume - - if display_rev: - if last_applied is None: - msg = ( - f"Checkpoint '{display_rev}' cannot be resumed until " - "it is applied to your workspace." - ) - else: - msg = ( - f"Checkpoint '{display_rev}' does not match the " - "most recently applied experiment in your workspace " - f"('{last_applied[:7]}')." - ) - msg = ( - f"{msg}\n" - "To resume this experiment run:\n\n" - f"\tdvc exp apply {display_rev}\n\n" - "And then retry this 'dvc exp res' command." - ) - else: - msg = "No existing checkpoint to resume in your workspace." - - raise DvcException(msg) - baseline_rev = self._get_baseline(branch) logger.debug( "Resume from checkpoint '%s' with baseline '%s'", - checkpoint_resume, + resume_rev, baseline_rev, ) @@ -473,11 +458,33 @@ class Experiments: **kwargs, ) - def _get_last_checkpoint(self) -> str: - rev = self.scm.get_ref(EXEC_CHECKPOINT) - if rev: - return rev - raise DvcException("No existing checkpoint experiment to continue") + def _get_last_checkpoint(self) -> Optional[str]: + try: + last_checkpoint = self.scm.get_ref(EXEC_CHECKPOINT) + if last_checkpoint: + self.check_baseline(last_checkpoint) + return last_checkpoint + except BaselineMismatchError: + # If HEAD has moved since the the last checkpoint run, + # the specified checkpoint is no longer relevant + self.scm.remove_ref(EXEC_CHECKPOINT) + return None + + def _get_last_applied(self) -> Optional[str]: + try: + last_applied = self.scm.get_ref(EXEC_APPLY) + if last_applied: + self.check_baseline(last_applied) + return last_applied + except BaselineMismatchError: + # If HEAD has moved since the the last applied experiment, + # the applied experiment is no longer relevant + self.scm.remove_ref(EXEC_APPLY) + return None + + def reset_checkpoints(self): + self.scm.remove_ref(EXEC_CHECKPOINT) + self.scm.remove_ref(EXEC_APPLY) @scm_locked def _reproduce_revs( diff --git a/dvc/repo/experiments/run.py b/dvc/repo/experiments/run.py index b96c7ab34..3851a0a59 100644 --- a/dvc/repo/experiments/run.py +++ b/dvc/repo/experiments/run.py @@ -12,9 +12,10 @@ def run( repo, targets: Optional[Iterable[str]] = None, params: Optional[Iterable[str]] = None, - run_all: Optional[bool] = False, - jobs: Optional[int] = 1, - tmp_dir: Optional[bool] = False, + run_all: bool = False, + jobs: int = 1, + tmp_dir: bool = False, + reset: bool = False, **kwargs, ) -> dict: """Reproduce the specified targets as an experiment. @@ -24,6 +25,10 @@ def run( Returns a dict mapping new experiment SHAs to the results of `repro` for that experiment. """ + if reset: + repo.experiments.reset_checkpoints() + kwargs["force"] = True + if run_all: return repo.experiments.reproduce_queued(jobs=jobs) diff --git a/tests/func/experiments/conftest.py b/tests/func/experiments/conftest.py index ffafce6d5..8a8b00c53 100644 --- a/tests/func/experiments/conftest.py +++ b/tests/func/experiments/conftest.py @@ -9,7 +9,6 @@ CHECKPOINT_SCRIPT_FORMAT = dedent( import os import sys import shutil - from time import sleep from dvc.api import make_checkpoint diff --git a/tests/func/experiments/test_checkpoints.py b/tests/func/experiments/test_checkpoints.py index 86dd92562..a6d01a11d 100644 --- a/tests/func/experiments/test_checkpoints.py +++ b/tests/func/experiments/test_checkpoints.py @@ -1,10 +1,8 @@ -import logging - import pytest from funcy import first from dvc.exceptions import DvcException -from dvc.repo.experiments import Experiments, MultipleBranchError +from dvc.repo.experiments import MultipleBranchError from dvc.repo.experiments.base import EXEC_APPLY, EXEC_CHECKPOINT @@ -38,23 +36,11 @@ def test_new_checkpoint( @pytest.mark.parametrize( "checkpoint_resume, workspace", - [ - (Experiments.LAST_CHECKPOINT, True), - (Experiments.LAST_CHECKPOINT, False), - ("foo", True), - ("foo", False), - ], + [(None, True), (None, False), ("foo", True), ("foo", False)], ) def test_resume_checkpoint( tmp_dir, scm, dvc, checkpoint_stage, checkpoint_resume, workspace ): - with pytest.raises(DvcException): - dvc.experiments.run( - checkpoint_stage=checkpoint_stage.addressing, - checkpoint_resume=checkpoint_resume, - tmp_dir=not workspace, - ) - results = dvc.experiments.run( checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace ) @@ -66,7 +52,7 @@ def test_resume_checkpoint( tmp_dir=not workspace, ) - if checkpoint_resume != Experiments.LAST_CHECKPOINT: + if checkpoint_resume: checkpoint_resume = first(results) if not workspace: @@ -96,8 +82,6 @@ def test_resume_checkpoint( def test_reset_checkpoint( tmp_dir, scm, dvc, checkpoint_stage, caplog, workspace ): - from dvc.repo.experiments.base import CheckpointExistsError - dvc.experiments.run( checkpoint_stage.addressing, name="foo", tmp_dir=not workspace, ) @@ -106,33 +90,12 @@ def test_reset_checkpoint( scm.reset(hard=True) scm.gitpython.repo.git.clean(force=True) - if workspace: - with caplog.at_level(logging.ERROR): - dvc.experiments.run( - checkpoint_stage.addressing, - name="foo", - params=["foo=2"], - tmp_dir=not workspace, - ) - assert "checkpoint experiment conflicts with existing" in caplog.text - - scm.reset(hard=True) - scm.gitpython.repo.git.clean(force=True) - else: - with pytest.raises(CheckpointExistsError): - dvc.experiments.run( - checkpoint_stage.addressing, - name="foo", - params=["foo=2"], - tmp_dir=not workspace, - ) - results = dvc.experiments.run( checkpoint_stage.addressing, params=["foo=2"], name="foo", - force=True, tmp_dir=not workspace, + reset=True, ) exp = first(results) @@ -165,16 +128,6 @@ def test_resume_branch(tmp_dir, scm, dvc, checkpoint_stage, workspace): tmp_dir=not workspace, ) checkpoint_a = first(results) - if not workspace: - dvc.experiments.apply(checkpoint_a, force=True) - - with pytest.raises(DvcException): - results = dvc.experiments.run( - checkpoint_stage.addressing, - checkpoint_resume=branch_rev, - params=["foo=100"], - tmp_dir=not workspace, - ) dvc.experiments.apply(branch_rev, force=True) results = dvc.experiments.run( diff --git a/tests/unit/command/test_experiments.py b/tests/unit/command/test_experiments.py index 1311873f5..c9865430c 100644 --- a/tests/unit/command/test_experiments.py +++ b/tests/unit/command/test_experiments.py @@ -84,19 +84,19 @@ def test_experiments_show(dvc, scm, mocker): @pytest.mark.parametrize( - "args, resume", [(["exp", "run"], None), (["exp", "resume"], ":last")] + "args, reset", [(["exp", "run"], False), (["exp", "reset"], True)], ) -def test_experiments_run(dvc, scm, mocker, args, resume): +def test_experiments_run(dvc, scm, mocker, args, reset): default_arguments = { "params": [], "name": None, "queue": False, "run_all": False, "jobs": None, - "checkpoint_resume": resume, "tmp_dir": False, + "checkpoint_resume": None, + "reset": reset, } - default_arguments.update(repro_arguments) cmd = CmdExperimentsRun(parse_args(args))