From aec8d657912b7fd2b4a0a13afc220d9496a1b999 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 16 Dec 2020 03:25:18 +0200 Subject: [PATCH] repro: accept multiple targets (#5111) Currently, we iterate over targets manually in CLI, which results in a complicated code and a DAG collection/check overhead. With this PR we collect the DAG once, figure out all the needed targets within it and just reproduce them in one pass. --- dvc/command/base.py | 8 --- dvc/command/experiments.py | 34 +++-------- dvc/command/repro.py | 40 ++++-------- dvc/repo/experiments/run.py | 8 +-- dvc/repo/reproduce.py | 84 ++++++++++++++------------ tests/func/test_repro_multistage.py | 4 +- tests/unit/command/test_experiments.py | 5 +- tests/unit/command/test_repro.py | 6 +- 8 files changed, 77 insertions(+), 112 deletions(-) diff --git a/dvc/command/base.py b/dvc/command/base.py index f1f2facb3..c70e99ad9 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -45,14 +45,6 @@ class CmdBase(ABC): updater = Updater(self.repo.tmp_dir, hardlink_lock=hardlink_lock) updater.check() - @property - def default_targets(self): - """Default targets for `dvc repro`.""" - from dvc.dvcfile import PIPELINE_FILE - - logger.trace(f"assuming default target '{PIPELINE_FILE}'.") - return [PIPELINE_FILE] - @abstractmethod def run(self): pass diff --git a/dvc/command/experiments.py b/dvc/command/experiments.py index bb1585c44..5c29f3f53 100644 --- a/dvc/command/experiments.py +++ b/dvc/command/experiments.py @@ -441,31 +441,17 @@ class CmdExperimentsDiff(CmdBase): class CmdExperimentsRun(CmdRepro): def run(self): - # Dirty hack so the for loop below can at least enter once - if self.args.all_pipelines: - self.args.targets = [None] - elif not self.args.targets: - self.args.targets = self.default_targets + self.repo.experiments.run( + name=self.args.name, + queue=self.args.queue, + run_all=self.args.run_all, + jobs=self.args.jobs, + params=self.args.params, + checkpoint_resume=self.args.checkpoint_resume, + **self._repro_kwargs, + ) - ret = 0 - for target in self.args.targets: - try: - self.repo.experiments.run( - target, - name=self.args.name, - queue=self.args.queue, - run_all=self.args.run_all, - jobs=self.args.jobs, - params=self.args.params, - checkpoint_resume=self.args.checkpoint_resume, - **self._repro_kwargs, - ) - except DvcException: - logger.exception("") - ret = 1 - break - - return ret + return 0 class CmdExperimentsGC(CmdRepro): diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 5e3ef6968..5b3374172 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -6,46 +6,30 @@ from dvc.command.base import CmdBase, append_doc_link from dvc.command.metrics import _show_metrics from dvc.command.status import CmdDataStatus from dvc.dvcfile import PIPELINE_FILE -from dvc.exceptions import DvcException logger = logging.getLogger(__name__) class CmdRepro(CmdBase): def run(self): - # Dirty hack so the for loop below can at least enter once - if self.args.all_pipelines: - self.args.targets = [None] - elif not self.args.targets: - self.args.targets = self.default_targets + stages = self.repo.reproduce(**self._repro_kwargs) + if len(stages) == 0: + logger.info(CmdDataStatus.UP_TO_DATE_MSG) + else: + logger.info( + "Use `dvc push` to send your updates to " "remote storage." + ) - ret = 0 - for target in self.args.targets: - try: - stages = self.repo.reproduce(target, **self._repro_kwargs) + if self.args.metrics: + metrics = self.repo.metrics.show() + logger.info(_show_metrics(metrics)) - if len(stages) == 0: - logger.info(CmdDataStatus.UP_TO_DATE_MSG) - else: - logger.info( - "Use `dvc push` to send your updates to " - "remote storage." - ) - - if self.args.metrics: - metrics = self.repo.metrics.show() - logger.info(_show_metrics(metrics)) - - except DvcException: - logger.exception("") - ret = 1 - break - - return ret + return 0 @property def _repro_kwargs(self): return { + "targets": self.args.targets, "single_item": self.args.single_item, "force": self.args.force, "dry": self.args.dry, diff --git a/dvc/repo/experiments/run.py b/dvc/repo/experiments/run.py index 8a9e21f81..54b43bc1f 100644 --- a/dvc/repo/experiments/run.py +++ b/dvc/repo/experiments/run.py @@ -37,13 +37,13 @@ def _parse_params(path_params: Iterable): @locked def run( repo, - target: Optional[str] = None, + targets: Optional[Iterable] = None, params: Optional[Iterable] = None, run_all: Optional[bool] = False, jobs: Optional[int] = 1, **kwargs, ) -> dict: - """Reproduce the specified target as an experiment. + """Reproduce the specified targets as an experiment. Accepts the same additional kwargs as Repo.reproduce. @@ -59,10 +59,10 @@ def run( params = [] try: return repo.experiments.reproduce_one( - target=target, params=params, **kwargs + targets=targets, params=params, **kwargs ) except UnchangedExperimentError: # If experiment contains no changes, just run regular repro kwargs.pop("queue", None) kwargs.pop("checkpoint_resume", None) - return {None: repo.reproduce(target=target, **kwargs)} + return {None: repo.reproduce(targets=targets, **kwargs)} diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py index b5385e1c7..54464ac05 100644 --- a/dvc/repo/reproduce.py +++ b/dvc/repo/reproduce.py @@ -3,7 +3,7 @@ import os import typing from functools import partial -from dvc.exceptions import InvalidArgumentError, ReproductionError +from dvc.exceptions import ReproductionError from dvc.repo.scm_context import scm_context from dvc.stage.run import CheckpointKilledError @@ -92,7 +92,7 @@ def _get_active_graph(G): @scm_context def reproduce( self: "Repo", - target=None, + targets=None, recursive=False, pipeline=False, all_pipelines=False, @@ -100,11 +100,14 @@ def reproduce( ): glob = kwargs.pop("glob", False) accept_group = not glob - assert target is None or isinstance(target, str) - if not target and not all_pipelines: - raise InvalidArgumentError( - "Neither `target` nor `--all-pipelines` are specified." - ) + + if isinstance(targets, str): + targets = [targets] + + if not all_pipelines and targets is None: + from dvc.dvcfile import PIPELINE_FILE + + targets = [PIPELINE_FILE] interactive = kwargs.get("interactive", False) if not interactive: @@ -113,28 +116,33 @@ def reproduce( active_graph = _get_active_graph(self.graph) active_pipelines = get_pipelines(active_graph) + stages = set() if pipeline or all_pipelines: if all_pipelines: pipelines = active_pipelines else: - stage = self.stage.get_target(target) - pipelines = [get_pipeline(active_pipelines, stage)] + pipelines = [] + for target in targets: + stage = self.stage.get_target(target) + pipelines.append(get_pipeline(active_pipelines, stage)) - targets = [] for pipeline in pipelines: for stage in pipeline: if pipeline.in_degree(stage) == 0: - targets.append(stage) + stages.add(stage) else: - targets = self.stage.collect( - target, - recursive=recursive, - graph=active_graph, - accept_group=accept_group, - glob=glob, - ) + for target in targets: + stages.update( + self.stage.collect( + target, + recursive=recursive, + graph=active_graph, + accept_group=accept_group, + glob=glob, + ) + ) - return _reproduce_stages(active_graph, targets, **kwargs) + return _reproduce_stages(active_graph, list(stages), **kwargs) def _reproduce_stages( @@ -220,30 +228,28 @@ def _reproduce_stages( def _get_pipeline(G, stages, downstream, single_item): import networkx as nx - if single_item: - all_pipelines = stages - else: - all_pipelines = [] - for stage in stages: - if downstream: - # NOTE (py3 only): - # Python's `deepcopy` defaults to pickle/unpickle the object. - # Stages are complex objects (with references to `repo`, - # `outs`, and `deps`) that cause struggles when you try - # to serialize them. We need to create a copy of the graph - # itself, and then reverse it, instead of using - # graph.reverse() directly because it calls `deepcopy` - # underneath -- unless copy=False is specified. - nodes = nx.dfs_postorder_nodes( - G.copy().reverse(copy=False), stage - ) - all_pipelines += reversed(list(nodes)) - else: - all_pipelines += nx.dfs_postorder_nodes(G, stage) + all_pipelines = [] + for stage in stages: + if downstream: + # NOTE (py3 only): + # Python's `deepcopy` defaults to pickle/unpickle the object. + # Stages are complex objects (with references to `repo`, + # `outs`, and `deps`) that cause struggles when you try + # to serialize them. We need to create a copy of the graph + # itself, and then reverse it, instead of using + # graph.reverse() directly because it calls `deepcopy` + # underneath -- unless copy=False is specified. + nodes = nx.dfs_postorder_nodes(G.copy().reverse(copy=False), stage) + all_pipelines += reversed(list(nodes)) + else: + all_pipelines += nx.dfs_postorder_nodes(G, stage) pipeline = [] for stage in all_pipelines: if stage not in pipeline: + if single_item and stage not in stages: + continue + pipeline.append(stage) return pipeline diff --git a/tests/func/test_repro_multistage.py b/tests/func/test_repro_multistage.py index 743c8765a..1885482d3 100644 --- a/tests/func/test_repro_multistage.py +++ b/tests/func/test_repro_multistage.py @@ -528,7 +528,7 @@ def test_repro_list_of_commands_in_order(tmp_dir, dvc): """ ) ) - dvc.reproduce(target="multi") + dvc.reproduce(targets=["multi"]) assert (tmp_dir / "foo").read_text() == "foo\n" assert (tmp_dir / "bar").read_text() == "bar\n" @@ -547,6 +547,6 @@ def test_repro_list_of_commands_raise_and_stops_after_failure(tmp_dir, dvc): ) ) with pytest.raises(ReproductionError): - dvc.reproduce(target="multi") + dvc.reproduce(targets=["multi"]) assert (tmp_dir / "foo").read_text() == "foo\n" assert not (tmp_dir / "bar").exists() diff --git a/tests/unit/command/test_experiments.py b/tests/unit/command/test_experiments.py index 0d8434806..59aed9cbf 100644 --- a/tests/unit/command/test_experiments.py +++ b/tests/unit/command/test_experiments.py @@ -12,7 +12,6 @@ from dvc.command.experiments import ( CmdExperimentsRun, CmdExperimentsShow, ) -from dvc.dvcfile import PIPELINE_FILE from dvc.exceptions import InvalidArgumentError from .test_repro import default_arguments as repro_arguments @@ -103,9 +102,7 @@ def test_experiments_run(dvc, scm, mocker, args, resume): mocker.patch.object(cmd.repo, "reproduce") mocker.patch.object(cmd.repo.experiments, "run") cmd.run() - cmd.repo.experiments.run.assert_called_with( - PIPELINE_FILE, **default_arguments - ) + cmd.repo.experiments.run.assert_called_with(**default_arguments) def test_experiments_gc(dvc, scm, mocker): diff --git a/tests/unit/command/test_repro.py b/tests/unit/command/test_repro.py index 172893b37..75fabbfe6 100644 --- a/tests/unit/command/test_repro.py +++ b/tests/unit/command/test_repro.py @@ -1,6 +1,5 @@ from dvc.cli import parse_args from dvc.command.repro import CmdRepro -from dvc.dvcfile import PIPELINE_FILE default_arguments = { "all_pipelines": False, @@ -16,6 +15,7 @@ default_arguments = { "force_downstream": False, "pull": False, "glob": False, + "targets": [], } @@ -23,7 +23,7 @@ def test_default_arguments(dvc, mocker): cmd = CmdRepro(parse_args(["repro"])) mocker.patch.object(cmd.repo, "reproduce") cmd.run() - cmd.repo.reproduce.assert_called_with(PIPELINE_FILE, **default_arguments) + cmd.repo.reproduce.assert_called_with(**default_arguments) def test_downstream(dvc, mocker): @@ -32,4 +32,4 @@ def test_downstream(dvc, mocker): cmd.run() arguments = default_arguments.copy() arguments.update({"downstream": True}) - cmd.repo.reproduce.assert_called_with(PIPELINE_FILE, **arguments) + cmd.repo.reproduce.assert_called_with(**arguments)