Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 07:18:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

repro: accept multiple targets (bp #5111) (#5115)

Browse Source 
* 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.

(cherry picked from commit aec8d65791)

# Conflicts:
#	dvc/command/base.py
#	dvc/command/experiments.py
#	dvc/command/repro.py
#	tests/func/test_repro_multistage.py

* resolve conflicts

* resolve test conflicts

Co-authored-by: Ruslan Kuprieiev <[email protected]>
Co-authored-by: Ruslan Kuprieiev <[email protected]>
This commit is contained in:
mergify[bot]
2020-12-16 05:17:55 +02:00
committed by GitHub
co-authored by Ruslan Kuprieiev Ruslan Kuprieiev
parent 4082757aa1
commit 29db8a1a95
7 changed files with 75 additions and 110 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/command/base.py
-8
View File
@@ -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.debug(f"assuming default target '{PIPELINE_FILE}'.")
return [PIPELINE_FILE]
@abstractmethod
def run(self):
pass
dvc/command/experiments.py
+10 -24
View File
@@ -451,32 +451,18 @@ class CmdExperimentsRun(CmdRepro):
saved_dir = os.path.realpath(os.curdir)
os.chdir(self.args.cwd)
# 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
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
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,
)
os.chdir(saved_dir)
return ret
return 0
class CmdExperimentsGC(CmdRepro):
dvc/command/repro.py
+12 -28
View File
@@ -7,7 +7,6 @@ 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__)
@@ -17,40 +16,25 @@ class CmdRepro(CmdBase):
saved_dir = os.path.realpath(os.curdir)
os.chdir(self.args.cwd)
# 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 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
if self.args.metrics:
metrics = self.repo.metrics.show()
logger.info(_show_metrics(metrics))
os.chdir(saved_dir)
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,
dvc/repo/experiments/run.py
+4 -4
View File
@@ -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)}
dvc/repo/reproduce.py
+45 -39
View File
@@ -2,7 +2,7 @@ import logging
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
@@ -80,7 +80,7 @@ def _get_active_graph(G):
@scm_context
def reproduce(
self: "Repo",
target=None,
targets=None,
recursive=False,
pipeline=False,
all_pipelines=False,
@@ -88,11 +88,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:
@@ -101,28 +104,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(
@@ -208,30 +216,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
tests/unit/command/test_experiments.py
+1 -4
View File
@@ -9,7 +9,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
@@ -100,9 +99,7 @@ def test_experiments_run(dvc, 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, mocker):
tests/unit/command/test_repro.py
+3 -3
View File
@@ -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)